Restructuring rows — reshape & cull

The question for this lesson: the last two lessons brought rows together: folding many into a summary (aggregate), joining two streams (combine), stacking peers (merge). But sometimes you don’t want to combine or summarize rows at all. You want to rewrite some of them (split one over-long record into two), or drop or route out whole groups of them. How do you do that?

Two node types handle this, and they work within a group, not across streams:

• reshape rewrites rows by rules inside a partition (e.g. close one record at a boundary and synthesize a continuation row).
• cull drops or side-routes whole groups by a group predicate (e.g. send every employee with more than three plan rows to a review output).

Run in a clinker checkout

These nodes work on the employee_plans data, which the in-page sandbox doesn’t carry, and they’re more involved than the customer examples. So this lesson teaches from Clinker’s real example pipelines and asks you to predict their output. Run them for real in a clinker checkout with clinker run … --dry-run / --explain.

What you’ll be able to do

• Use a reshape node to rewrite rows by rules within a partition (partition_by, order_by, rules, when, mutate.set, synthesize).
• Use a cull node to drop or side-route rows by a group predicate, and read the removed rows off its removed_to side-port.
• Explain that reshape is blocking and grouped (it sees a partition’s whole group before any rule fires, and its rules don’t cascade), and that cull’s removed rows go to a side-port, not silently away.

Reshape: rewrite rows by rules within a partition

A reshape node takes one input and rewrites rows according to rules, working within a partition (a group of related rows). This is Clinker’s real employee_plan_backfill reshape. Within each employee’s plan history, a coverage record that runs longer than a fiscal year is an un-split record, so reshape closes it at the start boundary and synthesizes a continuation:

- type: reshape
  name: backfill
  input: plans
  config:
    partition_by: [employee_id]          # the group: one employee's plan history
    order_by:
      - { field: plan_start, order: asc } # sort within the group, oldest first
    rules:
      - name: split_long_plan
        when: "plan_start - plan_end > 365"   # which rows this rule fires on (a CXL predicate)
        mutate:
          set:
            # Close the over-long window at its start boundary.
            plan_end: "plan_start"            # rewrite a field on the triggering row
        synthesize:
          copy_from: trigger                  # start the new row as a copy of the trigger row
          overrides:
            status: "'synthesized'"           # mark the continuation row

🌱 New here? — partition_by and order_by

reshape doesn’t look at one row at a time. partition_by: [employee_id] collects all the rows for an employee into one group, and order_by sorts that group (here, by plan_start, ascending). Only then do the rules run, so a rule can reason about a row’s place within its group, not just the row on its own.

The pieces of a rule:

• when: is a CXL predicate that decides which rows the rule fires on. Here, plan_start - plan_end > 365 picks the over-long records and leaves the rest untouched.
• mutate.set: rewrites fields on the triggering row. plan_end: "plan_start" closes the window at its start boundary. The value on the right is a CXL expression.
• synthesize: optionally emits an extra row alongside the mutated one. copy_from: trigger starts the new row as a copy of the row that fired, then overrides: change specific fields (here, status: "'synthesized'", where the inner quotes make it the literal string synthesized, not a field reference). The scd_type2 example uses copy_from: none instead, building the continuation row field by field.

Three idiosyncrasies make reshape behave unlike a plain transform:

• It is blocking and grouped. Like an aggregate, it holds rows back: it observes a partition’s whole group before any rule fires, so a rule can see the full picture for that employee.
• Rules don’t cascade. A rule’s output (the mutated row, the synthesized row) is not fed back through the rules. Each rule sees the original group, so you never get a chain reaction where one rule’s change triggers another.
• It can spill to disk under a memory budget. Because it buffers each group, reshape governs that buffer against the engine’s memory budget. The scd_type2 example sets pipeline.memory: { limit: "16K", backpressure: spill } to force the spill path even on a tiny fixture; the result is identical whether a group stayed in memory or round-tripped through disk.

✋ Predict first

Within employee E001's group, reshape sees two plan rows: { plan_start: 1000, plan_end: 100, status: 'baseline' } and { plan_start: 100, plan_end: 90, status: 'baseline' }. The rule is when: plan_start - plan_end > 365, mutate.set: { plan_end: plan_start }, and synthesize: { copy_from: trigger, overrides: { status: 'synthesized' } }. Which row fires, and how many rows leave reshape in total?

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

The first row fires: plan_start - plan_end is 1000 - 100 = 900, which is > 365. Two things happen to a firing row.

1. The triggering row is mutated. mutate.set closes the window by setting plan_end to plan_start, so it leaves as { plan_start: 1000, plan_end: 1000, status: baseline }.
2. One extra row is synthesized. copy_from: trigger copies the row that fired (its original values), then overrides sets status to the literal synthesized, giving { plan_start: 1000, plan_end: 100, status: synthesized }.

So that one firing row becomes two output rows.

The second row does not fire: 100 - 90 = 10, which is not > 365, so it passes through unchanged as a single row. (Read the predicate against the real columns: plan_start is the later boundary, so plan_start - plan_end measures how over-long a window is, and only the 1000 / 100 window is wide enough to split.) And because rules don’t cascade, the synthesized row is never itself re-tested. E001’s two input rows leave reshape as three: one unchanged, one mutated, and one synthesized.

Cull: drop or side-route whole groups by a group predicate

A cull node also works within a partition, but instead of rewriting rows it removes whole groups, and the removed rows are not thrown away. This is the employee_plan_backfill cull: after backfill, any employee with more than three plan rows is routed to a manual-review output:

- type: cull
  name: flag_large_histories
  input: backfill
  config:
    partition_by: [employee_id]          # group the rows by employee
    removed_to: review                   # name the SIDE-OUTPUT port for removed rows
    rules:
      - name: too_many_plans
        drop_group_when: "count(*) > 3"  # a GROUP-level predicate over the partition

A downstream node reads the removed rows by referencing the side-port, flag_large_histories.review:

# Main output: employees with a manageable plan history.
- type: output
  name: out
  input: flag_large_histories          # the rows that were KEPT

# Side output: employees flagged for manual review.
- type: output
  name: review
  input: flag_large_histories.review   # the rows that were REMOVED

The pieces:

• partition_by: is the same as reshape: it groups the rows so the predicate can be evaluated over a whole group, not one row.
• drop_group_when: is a group-level predicate. count(*) > 3 counts the rows in the partition; if the whole group satisfies it, the entire group is removed from the main stream.
• removed_to: names a side-output port. Removed rows are valid records deliberately partitioned onto a second stream; they go to <cull_name>.<removed_to>, where a downstream node can pick them up. This is not the dead-letter queue (that’s for errors, lesson 07). These are good rows, just sent down a different road.

🌱 New here? — a group-level predicate

drop_group_when: is evaluated over the group, using the same group aggregates you met with aggregate in lesson 05: count, sum, min, max, avg, collect. So count(*) > 3 means “this partition has more than three rows.” There is no bare any() aggregate, so “any row in the group matches X” must be written as a count, e.g. sum(if <cond> then 1 else 0) > 0.

✋ Predict first

cull has partition_by: [employee_id], removed_to: review, and drop_group_when: count(*) > 3. Employee E1 has 5 plan rows; employee E2 has 2 plan rows. After the cull, which rows reach the `out` output, and which reach the `review` output?

Decide on your answer before you open the reveal — committing a guess is what makes it stick.

Reveal the answer

E1’s group has count(*) = 5, which is > 3, so the whole E1 group (all 5 rows) is removed from the main stream and lands on the review side-port. E2’s group has count(*) = 2, which is not > 3, so all of E2’s rows stay on the main stream and reach out.

The key idea: cull works at the group level. It keeps or removes a partition as a unit, it doesn’t pick individual rows. And nothing is lost: E1’s rows are recoverable at flag_large_histories.review. (If you want to drop individual rows by a per-row condition, that’s CXL filter in a transform, lesson 02, not cull.)

Retrieval checkpoint

✓ Checkpoint

// quick check

From lesson 05: a cull's drop_group_when uses a GROUP predicate over the partition. Which functions can it use, and how do you express 'any row in the group matches a condition'?

• count/sum/min/max/avg/collect, and 'any row matches' is written as sum(if <cond> then 1 else 0) > 0, because there is no bare any() aggregate
• any() and all() are built in, so you just write any(<cond>)
• Only count(*) is allowed in a group predicate

// quick check

A reshape rule fires on a row and both mutates it and synthesizes a continuation row. Could the synthesized row trigger the same rule a second time?

• No, reshape rules don't cascade; a rule's output isn't re-fed through the rules
• Yes, every emitted row is re-tested against all rules until none fire
• Only if you set cascade: true on the rule

Common mistakes

• Expecting cull to silently drop rows. Removed rows aren’t gone; they go to the removed_to: side-port. Wire a downstream node to <cull_name>.<removed_to> (or you’ll wonder where they went).
• Expecting reshape rules to cascade. They don’t. Each rule sees the original group; a rule’s mutated or synthesized output is never re-fed through the rules.
• Reaching for a bare any() aggregate. It doesn’t exist. Use a count, e.g. sum(if <cond> then 1 else 0) > 0, in drop_group_when:.
• Treating cull like a per-row filter. cull keeps or removes a whole group. To drop individual rows by a per-row test, use CXL filter in a transform (lesson 02).

You should be able to:

• Use a reshape node with partition_by, order_by, and a rule (when + mutate.set + synthesize)
• Say why reshape is blocking, grouped, and non-cascading
• Use a cull node with partition_by, drop_group_when, and removed_to
• Read removed rows off the side-port (<cull_name>.<removed_to>)

Verify in the checkout:

clinker run employee_plan_backfill.yaml --dry-run
clinker run employee_plan_backfill.yaml --explain
clinker run scd_type2.yaml --explain

Go deeper (optional)

reshape rewrites rows within a group; cull removes whole groups (onto a side-port). Both are blocking and grouped: like aggregate, they hold a group back before acting, and they respect the memory budget by spilling to disk when a group gets large. Next: what happens when a row goes wrong, with error strategies and the dead-letter queue (which is a different thing from cull’s side-port: that’s for errors, this was for good rows you chose to route).