Module op 

Spec 06/07 operators implemented outside graph.rs — the operator fan-out seam.

Why this module exists

Operators are variants of the closed graph::Operator enum, and Graph dispatches fetch/push/input_schema/… over it. That enum + its match sites are a single file: if every operator’s logic also lived in graph.rs, N agents each implementing one operator would all edit the same file and collide.

The seam (the decision in the prep wave) keeps each operator’s struct and all its logic in its own file here (op/<name>.rs), and leaves only thin, stable wiring in graph.rs:

1. one Operator::<Name>(<Name>) enum variant,
2. a delegating arm in each dispatch fn — Op::<Name>(o) => o.fetch(self, req) / o.push(self, change),
3. a typed input_schema arm (pass-through for the non-reshaping operators),
4. an add_<name>(<Name>) -> NodeId builder that takes a fully-constructed value — so changing the struct’s fields never changes graph.rs.

The operator drives the graph through the pub(crate) Graph API: Graph::fetch, Graph::push, Graph::input_schema, and Graph::storage (stateful operators get a StorageId from Graph::alloc_storage).

Adding an operator (the template)

Skip is the worked example. To add Take/Cap/Exists/FlippedJoin/ UnionFanOut/UnionFanIn:

1. Copy skip.rs → op/<name>.rs; define the struct (fields private to the module — grow them freely) and impl its fetch/push (+ the filter-chain methods if it’s a chain link like Exists). Stateful operators store a StorageId and read state via g.storage(id).
2. Add the four wiring lines to graph.rs (variant + the two delegating arms + the input_schema arm; plus the set_output arm and add_<name> builder).
3. mod <name>; pub use <name>::<Name>; here.

Two agents adding two operators touch disjoint files plus a handful of distinct, auto-mergeable lines in graph.rs — not the same match arm. That is what makes the operator port fan out.

Structs

Cap

Cap: unordered top-N by PK-set membership, backed by a StorageId slot.

Exists

Exists: a relationship-size gate in the where filter sub-graph.

FlippedJoin

FlippedJoin: child-driven inner join (spec 06 §3.2). Two input ports (Port::JoinParent/Port::JoinChild) like Join; the child drives the fetch but the parent is the output row, so rel_slot resolves against the parent schema and input_schema is the parent’s.

Reduce

Reduce: an invertible aggregate, global or grouped. The running accumulator(s) live in Graph storage under Reduce::storage, not in the struct, so they survive across fetch/push.

Skip

Skip: a stateless, ordered operator that drops rows up to bound (the AST start). Stateless ⇒ no StorageId.

Take

Take: ordered, bounded top-N per partition, backed by a StorageId slot.

UnionFanIn

UnionFanIn (union-fan-in.ts): N branch inputs, one output. fetch k-way-merges the branch fetches with PK dedup (merge_node_streams). push either accumulates (during a fan-out broadcast — the owned model) or does a direct cross-branch dedup (a flipped child pushed while the fan-out is idle). On drain it collapses the accumulation via push_accumulated_changes and forwards the single result.

UnionFanOut

UnionFanOut (union-fan-out.ts): one input, N branches, built over the source. fetch delegates to the input; push broadcasts the change to every branch (each a filter pipeline and/or a FlippedJoin) then drives the paired UnionFanIn’s collapse. Unlike the filter FanOut it operates on full nodes via side-effecting Graph::push, not the return-based filter chain_push.

Enums

AggSpec

Which aggregate a Reduce column computes, in output-column order (REDUCE-DESIGN.md §5). All three are invertible — a Remove folds into a fixed-size accumulator without re-reading the inputs — which is what lets Reduce maintain them from a pure delta stream.