Cull Nodes

Cull nodes observe a whole correlation group and remove the entire group when a group-level predicate holds, routing the removed records to a side-output port instead of discarding them. They are the node for “look at everything an entity did, then set the whole entity aside for review or reprocessing” — work that operates on an aggregate property of a group, not on individual rows:

• Route fans out per record on a row predicate.
• Reshape mutates rows and synthesizes new ones within a group.
• Aggregate reduces a group to one summary row.

None of those remove a whole correlation group based on an aggregate property of the group and emit the removed rows on a second stream. Cull does.

Cull is a blocking grouping operator: it buffers every record of a group before any output leaves, because a group-level predicate (e.g. “this group has more than 100 rows”) cannot be decided until the whole group is seen. It has two output ports: the main port (kept groups) and the removed_to side-output port (removed groups).

Basic structure

- type: cull
  name: flag_large_histories
  input: backfill
  config:
    partition_by: [employee_id]
    removed_to: review
    rules:
      - name: too_many_plans
        drop_group_when: "count(*) > 3"

For each employee_id group, if the group holds more than three rows the whole group is routed to the review side output; every other group flows to the main output.

partition_by

A list of field names. Records sharing the same values for all partition_by fields form one group, and the removal predicate observes one whole group at a time. This is the correlation key the operator reasons over. partition_by must cover every visible correlation-key field so group identity is preserved on both output ports.

Whole-input grouping (partition_by: [])

An empty list is the degenerate case of that key: every record shares it, so the entire input forms one group and drop_group_when decides the whole dataset at once — every record is kept, or every record is routed to removed_to.

- type: cull
  name: drop_bad
  input: events
  config:
    partition_by: []          # one group: the whole input
    removed_to: removed
    rules:
      - name: drop_any_error
        drop_group_when: "sum(if status == 'error' then 1 else 0) > 0"

That pipeline routes all records to removed if any single record has status: error. Keyed by [account] the same rule would remove only the offending account’s records — so reach for whole-input grouping when the decision genuinely concerns the batch (an all-or-nothing gate on a delivery), not when you meant a per-entity rule.

The whole input must then fit memory.limit at finalize, since a single group has to be resident when its predicate runs (see the limit below). The sibling group-count bound does not apply: one group is as few as the decision state can be.

Values Cull cannot key

Cull groups a record under a single null group in two cases: the column is absent from the record, or its value is an explicit null.

Everything else is either its own group or a hard error:

• An empty string ("") is its own group, distinct from account=null.
• A NaN float, or an array- or map-valued cell, aborts the run rather than grouping — a partition key must be a single scalar value. This abort currently presents as an internal error, but it is a data condition, not an engine defect: fix the offending column rather than treating the message as an engine invariant failure.

Reshape treats both of those differently: it folds empty strings, NaNs, and multi-value cells all into its null group instead. The blank-versus-null divergence is tracked in #1022. The unkeyable-value behavior is separate; until both rules are deliberately aligned or retained, do not assume one node’s grouping matches the other’s.

order_by

Optional. A list of sort fields ({ field, order }, where order is asc or desc) applied within each group before its predicate runs, so an order-sensitive predicate is deterministic. Nulls sort last. Arrival order breaks ties.

Rules

Each entry in rules: is a declarative removal rule with a name and a drop_group_when predicate. A group is removed when any rule’s predicate holds (the rules are OR-combined).

drop_group_when — the group-level removal predicate

drop_group_when is a CXL boolean expression evaluated in aggregate context over the whole group (group-by = partition_by). Because it is an aggregate expression, it uses CXL’s aggregate functions:

rules:
  - name: too_many_plans
    drop_group_when: "count(*) > 3"
  - name: high_total
    drop_group_when: "sum(amount) > 10000"

CXL’s bare aggregate vocabulary is sum / count / min / max / avg / collect / weighted_avg — there is no bare any() aggregate. To express “remove the group if any row matches a condition”, sum an indicator and compare to zero:

rules:
  - name: drop_error_groups
    # Remove any account group containing at least one `error` row.
    drop_group_when: "sum(if status == 'error' then 1 else 0) > 0"

Ordered comparisons (>, <, >=, <=) work over every comparable aggregate type, not just numbers — the predicate uses the same comparison rules as a Transform. Numbers, strings, and dates all order:

rules:
  - name: late_alphabet
    drop_group_when: "max(name) > 'M'"          # string ordering
  - name: recent_hire
    drop_group_when: "max(hired) >= #2020-01-01#" # date ordering (`#YYYY-MM-DD#` literal)

A group whose aggregate operand is null (for example max(...) over an all-null column) compares as false — a null operand never removes the group and never errors.

Comments in a predicate

A drop_group_when predicate may carry a # line-comment to explain the rule inline. Each rule’s predicate is parsed on its own, so a trailing comment applies only to that rule:

rules:
  - name: drop_error_groups
    drop_group_when: "sum(if status == 'error' then 1 else 0) > 0  # any error row removes the group"
  - name: high_total
    drop_group_when: "sum(amount) > 10000                          # large accounts"

The comment is source text only — it never changes the compiled decision. (A #YYYY-MM-DD# date literal is unaffected: it lexes as a date, not a comment.)

Output ports: main and removed_to

Cull has two producer-side output ports, the same mechanism a Route uses for its branches — not the dead-letter queue. Removed records are valid rows the operator deliberately partitions onto a second stream, not errors.

Downstream nodes draw from the two ports by reference:

• The main output (kept groups) is referenced by the Cull node’s bare name: input: flag_large_histories.
• The side output (removed groups) is referenced as <cull>.<removed_to>: input: flag_large_histories.review.

  - type: output
    name: kept
    input: flag_large_histories            # main port — kept groups
    config: { name: kept, type: csv, path: kept.csv }

  - type: output
    name: review
    input: flag_large_histories.review     # side-output port — removed groups
    config: { name: review, type: csv, path: review.csv }

removed_to must be a non-empty name distinct from the Cull node’s own name (enforced at compile time, so the two ports are always distinguishable).

A single downstream node may draw from both ports — for example a Merge recombining the kept and removed streams (inputs: [flag_large_histories, flag_large_histories.review]) — and receives the union of both ports’ records.

removed_to is not the DLQ

The removed_to port carries the unchanged upstream schema — Cull does not widen, and both ports emit exactly the input columns. Removed records are not DlqEntrys and never appear in the dead-letter queue or its counters; they flow down a normal data edge to whatever node draws the removed_to port (an audit sink, a reprocessing branch, another transform). Use the DLQ for errors; use a Cull side output for valid records you want to handle separately. See Error Handling & DLQ for the error path.

Memory model

Cull is a blocking, grouped operator: it groups every input record by partition_by before any record leaves, because the group-level drop_group_when predicate is an aggregate property of the whole group and cannot be folded into a per-record keep/remove decision. It therefore cannot stream — the full group set materializes before the first output row leaves.

That per-group buffer is governed by the same central memory arbitrator every other blocking operator polls (see Memory & Spill). As records are grouped, Cull tracks the live in-memory footprint and, whenever the run crosses the soft spill threshold (80% of memory.limit by default), it spills buffered groups to disk:

• What spills: the raw input records. On reload at finalize, each group is re-split onto its output port exactly as it would have been without spilling, so the output is identical whether a group stayed resident or round-tripped through disk — including within-group row order, which is restored to arrival order after a reload. (The per-group removal decision is computed from an in-memory aggregate over the same records; that aggregate state is O(distinct groups) and is never spilled — only the raw records spill. It cannot spill because Cull has no upstream channel to pause, so instead it is bounded by a hard check: if that state plus the run’s other live charged memory would exceed memory.limit, the run fails loud with a memory-budget error rather than growing it unbounded. See the limit below.)
• Spill priority: 15, between grace-hash Combine (10) and external sort (20), matching Reshape. Cull cannot back-pressure — once its predecessor has drained there is no upstream channel to pause — so under memory pressure it always spills its own buffer in-thread rather than pausing a producer.
• Largest-first, stop at the threshold: when the budget trips, Cull evicts the largest resident groups first and stops as soon as the resident footprint drops back under the soft threshold — it does not drain every group.
• Skew (one giant group): a single correlation group whose resident tail alone exceeds the budget is spilled incrementally — sliced by the upper bits of each record’s admission sequence — while smaller groups stay resident, so the ingest-time resident peak stays bounded even under one giant skewed group.

The on-disk spill volume Cull produces is surfaced per stage in clinker run --explain (the Estimated spill volume and Spill compression sections) and, after a run, in the actual per-stage spill totals.

Limit: a single group must fit the finalize budget

Cull evaluates its group-level predicate against the whole group at once, so even though cross-group and ingest-time peaks spill to disk, the finalize reload of one group needs that group to fit the memory budget. Skew slicing bounds the ingest peak, but a single correlation group larger than memory.limit has no in-budget representation. Rather than risk an out-of-memory crash, the run fails loud with E310 MemoryBudgetExceeded, naming the Cull node, the offending partition_by group, and its footprint against the budget:

E310 drop_big: arena exceeded budget (512000/8192) [one Cull correlation
group [account="BIG"] does not fit; the reported use is that group's reload
footprint alone. ...]

Raising memory.limit is the only fix that leaves your output unchanged. Raise it clear of the reported figure — finalize also holds the run’s remaining groups and the per-group decision map, so that figure is a floor, not a target.

The other two levers both change what you get, and are worth knowing only so you can weigh them deliberately:

• Dropping columns this node does not read, in an upstream Transform, shrinks each buffered record. But Cull filters rows, never columns — both its ports carry the unchanged upstream schema — so anything you strip upstream is also missing from the main and removed_to outputs.
• Narrowing partition_by shrinks the group too, but that key defines the group drop_group_when evaluates over. With a rule like count(*) > 100, splitting one account across a finer key drops each resulting group below the threshold, so an account that should have been removed is emitted on the main port instead — the run “works” and quietly returns a different result set. Treat the grouping key as a modelling decision, never as a memory knob.

Run clinker explain --code E310 for remediation keyed to whichever memory surface overran, or see the memory guide.

There is a second, symmetric bound on the number of groups. The per-group removal decision is held in an in-memory aggregate that is O(distinct groups) and — unlike the raw records — cannot spill. If a partition key is so high-cardinality that the decision state plus the run’s other live charged memory would exceed memory.limit (many small groups rather than one giant group), the run likewise fails loud with E310, rather than growing that state unbounded. Here, coarsening partition_by is a legitimate fix only if the coarser key is the grouping you actually meant — the same caveat as above applies. Otherwise, raise memory.limit.

Several distinct memory surfaces can raise E310 against a Cull node, so read the [...] detail to see which one overran. The ones documented here are Cull correlation group [...] (the giant-group case above), Cull drop-decision aggregate state (the group-count bound), Cull cross-region tee admission (a downstream stage in a different deferred region forces this node’s output to be parked in memory), and node-buffer materialization overlap (a consumer must collect one of Cull’s sequential port scans into a resident vector). Cull’s main and removed_to handoff buffers themselves are spill-eligible, including when a port fans out. Other surfaces the shared runtime charges may name this node too — the detail string is the authority, not this list.