Operator catalogue¶

One table per group. The columns are:

Operator: the constructor, with its configuration arguments.
Signature: what __call__ takes and returns, then what compute_output_signature reports (None means "not tracked", as for Fanout and the parallel maps today).
Semantics: one line; the detail is in operators.md and execution.md.
Status: today (ships in pipekit-ops 0.2.1, file named), proposed (Pn) (the roadmap phase in ../roadmap.md that adds it), or dropped (not added, with the reason). Phases P1 to P5 have shipped, so every row that this design adds is now today; a "rebuilt on Map" note marks a class that shipped before and was rebuilt in P2.

Conventions shared by every new operator¶

Keys are plain values. Where an argument is a key (key, keys), it is a str, an int or a tuple of them, and it is used for item lookup (x[key]), as in toolz's getter (toolz/itertoolz.py, def getter). The per-item key of GroupBy, CountBy, Unique and SortBy may instead be an Operator, which is called on the item. As built, the dict operators take plain keys only, because Get(op) would just be op. A bare Python callable is refused with a TypeError that says to wrap it in Lambda.
Predicates and functions are operators. Filter(pred), Until(op, pred, …), AssertThat(pred) and the rest take an Operator. Their config is then the child's config, so they round-trip whenever their children do, and no new class sets forbid_in_yaml.
Lists in, lists out. Iterable operators accept any iterable and return a list (or a dict for the group-by family), as the parallel maps do today (packages/pipekit-ops/src/pipekit_ops/parallel.py). The two exceptions return a lazy iterator on purpose: Prefetch and Map.iter.
Several inputs, positionally. Merge and Zip take their inputs as positional arguments, so they work as Graph nodes (Merge()(a, b)), and also accept one sequence of inputs, so they work inside a Sequential after juxt or Fanout.

Dict plumbing¶

Graph and Fanout return dicts (packages/pipekit-ops/src/pipekit_ops/_base/graph.py, packages/pipekit-ops/src/pipekit_ops/combine.py), and STAC items are nested dicts. Before P1, only Lambda reshaped them.

Operator	Signature	Semantics	Status
`Get(key, default=MISSING)`	`Mapping \\| Sequence → Any`; `None`	`x[key]`; a tuple of keys returns a tuple; `default` replaces a missing key	today (`dicts.py`)
`GetIn(keys, default=MISSING)`	`Mapping → Any`; `None`	`x[k0][k1]…`; `default` replaces a missing path	today (`dicts.py`)
`Pick(keys)`	`Mapping → dict`; `None`	keep only `keys`, in the order given; a missing key raises `KeyError`	today (`dicts.py`)
`Dissoc(keys)`	`Mapping → dict`; `None`	drop `keys`; missing keys are ignored	today (`dicts.py`)
`Rename(mapping)`	`Mapping → dict`; `None`	rename keys by `mapping`; unlisted keys pass through; a rename onto an existing key raises `ValueError`	today (`dicts.py`)
`Merge()`	`*Mapping → dict`; `None`	union, later inputs win (toolz `merge`)	today (`dicts.py`)
`Assoc(key, op)`	`Mapping → dict`; `None`	add `key: op(x)` computed from the whole dict	today (`dicts.py`)
`UpdateIn(keys, op, default=MISSING)`	`Mapping → dict`; `None`	replace the value at a key path by `op(value)`; replaces the planned `OnKey`	today (`dicts.py`)
`ValMap(op)`	`Mapping → dict`; `None`	`op` on every value, keys kept (toolz `valmap`)	today (`dicts.py`)

Iterables¶

Operator	Signature	Semantics	Status
`Map(op, executor="serial", …)`	`Iterable → list`; `None`	`op` on every item; the execution engine, see execution.md	today (`execution.py`)
`Filter(pred)`	`Iterable → list`; `None`	keep items where `pred(item)` is truthy	today (`seq.py`)
`Not(pred)`	`Any → bool`; `None`	`not pred(x)`; the operator form of `complement`	today (`seq.py`)
`Remove(pred)`	`Iterable → list`; `None`	drop items where `pred(item)` is truthy (toolz `remove`)	today (`seq.py`)
`Reduce(op, initial=MISSING)`	`Iterable → Any`; `None`	left fold; `op` is called on the tuple `(acc, item)`; empty input with no `initial` raises `TypeError` like `functools.reduce`	today (`seq.py`)
`Fold(op, initial, chunksize=128, combine=None, executor="scope")`	`Iterable → Any`; `None`	associative tree fold over chunks, parallel through `Map` (toolz `sandbox.fold`)	today (`execution.py`)
`PartitionAll(n)`	`Iterable → list[tuple]`; `None`	consecutive tuples of length `n`, the last may be shorter; replaces the planned `Chunk`	today (`seq.py`)
`Concat()`	`Iterable[Iterable] → list`; `None`	flatten one level (toolz `concat`); replaces the planned `Flatten`	today (`seq.py`)
`Zip(strict=True)`	`*Iterable → list[tuple]`; `None`	builtin `zip`, strict by default	today (`seq.py`)
`GroupBy(key)`	`Iterable → dict[Any, list]`; `None`	group items by `key`, first-seen key order (toolz `groupby`)	today (`seq.py`)
`CountBy(key)`	`Iterable → dict[Any, int]`; `None`	count items per `key` (toolz `countby`; `CountBy(Identity())` is toolz `frequencies`)	today (`seq.py`)
`Count()`	`Iterable → int`; `None`	number of items, lazy iterables included (toolz `count`)	today (`seq.py`)
`Unique(key=None)`	`Iterable → list`; `None`	first occurrence of each item, or of each `key(item)` (toolz `unique`)	today (`seq.py`)
`Take(n)`	`Iterable → list`; `None`	the first `n` items (toolz `take`), for dry runs	today (`seq.py`)
`SortBy(key, reverse=False)`	`Iterable → list`; `None`	stable sort by `key`	today (`seq.py`)

Flow¶

Operator	Signature	Semantics	Status
`Retry(op, attempts, base_delay, on)`	`Any → Any`; `op`'s	exponential back-off, `base_delay * 2 ** (i - 1)` before attempt \(i+1\)	today (`control.py`)
`Try(primary, fallback, on)`	`Any → Any`; common of both	toolz `excepts`	today (`control.py`)
`Timeout(op, seconds, max_abandoned=8)`	`Any → Any`; `op`'s	raise `TimeoutError` if `op` has not returned after `seconds`; a daemon thread per call, abandoned, not killed; warns when more than `max_abandoned` are still running	today (`flow.py`)
`RateLimit(op, calls, period)`	`Any → Any`; `op`'s	at most `calls` starts of `op` in any half-open window of `period` seconds, per process	today (`flow.py`)
`Until(op, pred, max_iter, on_max="raise")`	`Any → Any`; `None`	apply `op` until `pred(x)` is truthy; stateless sibling of `pipekit_cycle.Recurrence`	today (`flow.py`)

Execution¶

Operator	Signature	Semantics	Status
`Map(op, executor="serial", n_workers=None, max_in_flight=None, on_error="raise", chunksize=1, start_method=None, progress=None)`	`Iterable → list`; `None`	one engine over serial, thread, process, async and any `concurrent.futures.Executor`; `progress` logs counts every `progress` items	today (`execution.py`)
`Map.iter(items, ordered=True)`, `await Map.run(items)`	lazy iterator; awaitable list	streaming and running-event-loop entry points; `ordered` lives here, not in the config	today (`execution.py`)
`MapError`, `Failed(index, exception)`	exception; record	the `ExceptionGroup` that `on_error="collect"` raises, and the record in each failed slot of `MapError.results`	today (`execution.py`)
`Prefetch(op, depth=2, executor="thread")`	`Iterable → Iterator`; `None`	read `depth` items ahead of the consumer, in input order	today (`execution.py`)
`executor_scope(executor)`	context manager	sets the executor that `Map(executor="scope")` uses (serial when none is set); not stored in any config	today (`execution.py`)
`ThreadMap(op, n_workers=8)`	`Iterable → list`; `None`	preset: `Map(op, executor="thread", …)`; config and hash unchanged	today (`parallel.py`), rebuilt on `Map`
`ProcessMap(op, n_workers=8, on_error="raise")`	`Iterable → list`; `None`	preset: `Map(op, executor="process", …)`; fails fast on `"raise"`; keeps the platform start method for one minor release, with a `FutureWarning` when it is not spawn	today (`parallel.py`), rebuilt on `Map`
`AsyncMap(op, semaphore=8)`	`Iterable → list`; `None`	preset: `Map(op, executor="async", max_in_flight=semaphore)`	today (`parallel.py`), rebuilt on `Map`
`BatchedMap(op, batch_size=8, flatten=True)`	`Iterable → list`; `None`	preset: serial `Map(op)` over lists of `batch_size` items, then an optional flatten; `PartitionAll` and the `executor` argument are still proposed	today (`parallel.py`), rebuilt on `Map`
`check_pickleable(op)`	`Operator → list[Operator]`	flagged operators in the tree	today (`parallel.py`); `Map` calls it, see execution.md

Cache and store¶

Operator	Signature	Semantics	Status
`Cache(inner, store=None, key=None)`	`Any → Any`; `inner`'s	memoise by input and the inner pipeline's hash; `store` names a `Store` in scope, `None` keeps today's private in-memory dict; `key` is an operator that computes the input key (toolz `memoize(key=…)`)	today (`cache.py`)
`Memoize`	alias	`Memoize is Cache`	today (`cache.py`)
`Store`	protocol	`__contains__(key)`, `load(key)`, `save(key, value)`	today (`store.py`)
`MemoryStore()`	`Store`	dict plus a lock; one process	today (`store.py`)
`DirectoryStore(root, codec="pickle")`	`Store`	one file per key, atomic rename, safe across processes on one file system; writes a `pipekit-store.json` format marker	today (`store.py`)
`store_scope(**stores)`	context manager	binds store names to `Store` instances; not stored in any config	today (`store.py`)
`Checkpoint(store)`	none	not added: it is `Cache(op, store=…)`, and `pipekit_train.Checkpoint` already owns the name	dropped

Observe and QC¶

Operator	Signature	Semantics	Status
`Tap`, `Snapshot`, `ShapeTrace`, `Profile`, `Histogram`	pass-through	closure-holding observers	today (`observe.py`)
`Log(message="call {n}: {type} {signature}", level="INFO", logger="pipekit_ops")`	`Any → Any`; passthrough	pass-through; logs one line per call with the call count and the input's `Signature` if an adapter is registered	today (`observe.py`)
`Map(…, progress=every)`	argument	log done / failed / in-flight counts and items per second every `every` items	today (`execution.py`)
`AssertThat(pred, message="predicate returned False")`	`Any → Any`; passthrough	`QCError` if `pred(x)` is falsy; the serialisable `AssertCallable`	today (`qc.py`)
`AssertSignature(pattern, dtype=None, missing="raise")`	`Any → Any`; passthrough	`QCError` unless `signature_of(x).matches(pattern)` (and the dtype agrees)	today (`qc.py`)
`Quarantine`, `AssertShape`, `AssertDType`, `AssertHasAttribute`, `AssertCallable`, `QCError`	pass-through	existing guards	today (`qc.py`)
`Progress`, observer `Count`	none	not added: `Map(progress=…)` and `Log`'s call count cover them, and `Count` keeps toolz's meaning	dropped

Name clash: two `BatchedMap`s¶

pipekit_ops.BatchedMap takes an iterable of items and calls the inner operator on lists of them (packages/pipekit-ops/src/pipekit_ops/parallel.py). The pipekit-array design plans a BatchedMap(inner, batch_size, axis) that splits one array along an axis (packages/pipekit-array/docs/design/api/operators.md, ADR A5 in packages/pipekit-array/docs/design/decisions.md). ADR A5 says the import path disambiguates. This design disagrees (decision O14): code that imports both packages, or a Hydra YAML that names both, meets one name with two meanings. The core name has shipped since v0.1, so the array operator should take the sibling proposal's name, BatchAlong(inner, batch_size, dim).