Add a transform/operator

A format plugs into an open dyn seam (adding a reader/writer format). An operator is the opposite: node kinds are a closed, engine-owned set (the wildcard-free PlanNode enum from dispatching a node), so adding one means editing the closed set itself rather than plugging into a seam. That makes this the change that touches the most places. The good news is that the engine’s two wildcard-free matches turn most of those places into compile errors that hand you the to-do list.

Prerequisites

• Add a reader/writer/format

→

Leads to

• Review & the gauntlet

What you’ll be able to do

• Trace the path from an authored YAML step (type:) through lowering and dispatch to a running operator body.
• Name the four edit sites adding an operator touches, and say which two the compiler forces on you.
• Predict what cargo check reports after you add both enum variants but neither match arm.
• Explain why the engine keeps two mirrored node enums (PipelineNode and PlanNode) instead of one shared enum.

New terms in this lesson (each is also expanded inline at the point you first need it):

• operator: one node kind in the pipeline DAG (source, transform, route, …). “Operator” and “node kind” name the same thing here, and adding one is this lesson’s task.
• transform: the specific operator that runs a CXL program on each record (type: transform in YAML). Used as the worked example of an operator that round-trips from YAML.
• PipelineNode: the authored node enum: one variant per YAML node type:, span-carrying, strictly deserialized. Lives on the config side (clinker-plan).
• PlanNode: the lowered node enum, the closed set the executor runs on (the wildcard-free enum from dispatching a node). A superset of PipelineNode, it also holds planner-synthesized kinds like Sort.
• lowering: the one-way match that turns each PipelineNode into a PlanNode at compile time (lower_node_to_plan_node), building the DAG. The bridge across the plan/runtime boundary.

Predict first

✋ Predict first

Adding a Dedup operator, you add a Dedup variant to BOTH node enums (PipelineNode, authored, and PlanNode, lowered), then run cargo check, having written no other code. Both lower_node_to_plan_node and dispatch_plan_node are wildcard-free matches (no _ => arm). Does cargo check (a) build clean and silently no-op the new node at run time, (b) report exactly one error, or (c) report exactly two errors? Write your pick and where the error(s) land before revealing.

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

Reveal the answer

The answer is (c): exactly two errors, one at lower_node_to_plan_node and one at dispatch_plan_node.

Each of those two matches enumerates one arm per node kind and has no _ => catch-all. Add a variant to the enum a match is over, and that match is now non-exhaustive: the compiler refuses to build it until you add the missing arm. So the two wildcard-free matches hand you a precise to-do list (“you still owe an arm here, and here”), naming exactly sites 3 and 4 of the four-place change.

The tempting wrong model is (a): “an unhandled node just falls through and does nothing at run time.” That is how an open dispatch design behaves: a registry lookup that finds no handler, a dyn table with a missing entry, a silent runtime no-op you discover in production. Clinker is built the other way (the closed-enum choice from dispatching a node): a forgotten operator is a compile error, never a silent drop. Hold that contrast. It is the whole reason this change is safe to make.

Two mirrored enums, one boundary

The engine represents a pipeline node twice, on two sides of the plan/runtime boundary you met in Planning & Expressions:

• PipelineNode: the authored node, as it appears in YAML (type: transform, type: route, …). It lives in the config layer (clinker-plan), carries spans, and is deserialized with strict per-variant checks.
• PlanNode: the lowered node, the executor’s vocabulary (the closed enum from dispatching a node). It carries resolved schemas, parallelism class, and other compile-time enrichment.

clinker-plan ·pipeline_node.rs ·PipelineNode type @19acdcb4

// The authored side — one variant per YAML node `type:`.
pub enum PipelineNode {
    Source { header: NodeHeader, config: SourceBody },
    Transform { header: NodeHeader, config: TransformBody },
    Aggregate { header: NodeHeader, config: AggregateBody },
    Route { header: NodeHeader, config: RouteBody },
    Merge { /* ... */ },
    // ...11 authored node kinds
}

clinker-plan ·mod.rs ·PlanNode type @19acdcb4

// The executor side — the closed set the DAG runs on.
pub enum PlanNode {
    Source { name: String, /* ... */ },
    Transform { name: String, /* resolved payload, parallelism, schema */ },
    Sort { name: String, /* ... */ sort_fields: Vec<SortField> },
    Aggregation { /* ... */ },
    // ...13 variants — a superset of PipelineNode
}

They are deliberately not identical. PipelineNode::Aggregate lowers to PlanNode::Aggregation (the names differ), and PlanNode has variants like Sort and CorrelationCommit that are planner-synthesized: injected during compilation, never authored in YAML, so they have no PipelineNode counterpart. Teaching tip: build your first operator as one that does round-trip from YAML (Transform, Route, Reshape, Cull) so you exercise the whole path.

The path from YAML to a running operator

 YAML  ──serde/strict──▶  PipelineNode::Route   (authored, in clinker-plan/config)
                                  │
                                  │  lower_node_to_plan_node   (the lowering match)
                                  ▼
                          PlanNode::Route        (lowered, goes into the DAG)
                                  │
                                  │  graph.add_node(...)  →  ExecutionPlanDag  →  CompiledPlan
                                  ▼
                          dispatch_plan_node      (the executor's exhaustive match)
                                  │
                                  ▼
                          dispatch_route → the operator body (clinker-exec)

The lowering match is one free function (config variant in, plan variant out) with no intermediate IR:

clinker-plan ·pipeline.rs ·lower_node_to_plan_node fn @19acdcb4

pub(crate) fn lower_node_to_plan_node(
    node: &PipelineNode,
    // ...plus the node's identity, span, and compile-time context
) -> Option<PlanNode> {
    match node {
        PipelineNode::Route { config, .. } => Some(PlanNode::Route {
            // the lowered payload — the route's branch config, resolved:
            mode: config.mode,
            branches: config.conditions.keys().cloned().collect(),
            default: config.default.clone(),
            // ...plus the per-node boilerplate (name, span, ...)
        }),
        PipelineNode::Aggregate { config, .. } => Some(PlanNode::Aggregation { /* ... */ }),
        // ...one arm per authored variant, no `_ =>`. Returns None when an
        // earlier stage already errored on this node.
    }
}

This runs inside the compile step (Stage 5), once per authored node, building the DAG:

// PipelineConfig::compile_with_diagnostics, Phase 1: one PlanNode per node.
for spanned in &self.nodes {
    let plan_node = lower_node_to_plan_node(node, /* node identity, span, compile ctx */);
    if let Some(pn) = plan_node {
        let idx = graph.add_node(pn);     // into DiGraph<PlanNode, PlanEdge>
        name_to_idx.insert(name, idx);
    }
}

The resulting DiGraph<PlanNode, _> becomes the ExecutionPlanDag inside CompiledPlan, the typed handle from the plan/runtime boundary. At run time the executor walks it and dispatches each node through the wildcard-free match you saw in dispatching a node:

clinker-exec ·dispatch.rs ·dispatch_plan_node fn @19acdcb4

match node {
    PlanNode::Source { .. }    => dispatch_source(ctx, current_dag, node_idx, &node)?,
    PlanNode::Transform { .. } => dispatch_transform(ctx, current_dag, node_idx, &node)?,
    PlanNode::Route { .. }     => dispatch_route(ctx, current_dag, node_idx, &node)?,
    // ...one arm per PlanNode variant, no `_ =>`
}

The four edit sites

Adding an operator means a coordinated change across four places, and crucially, two of them are wildcard-free matches that will not compile until you add the arm:

#	Site	File	Compiler-forced?
1	PipelineNode variant (+ Body struct, strict deserialize)	clinker-plan/src/config/pipeline_node.rs	— (you author it)
2	PlanNode variant	clinker-plan/src/plan/execution/mod.rs	— (you author it)
3	lower_node_to_plan_node arm + Phase-2 edge wiring	clinker-plan/src/config/pipeline.rs	yes (exhaustive match)
4	dispatch_plan_node arm → operator body	clinker-exec/src/executor/dispatch.rs (+ a *_dispatch.rs module)	yes (exhaustive match)

Sites 1–3 are all in clinker-plan (config, plan, lowering); site 4 is in clinker-exec (dispatch + the operator’s actual record logic). That split is the plan/runtime boundary again: planning lowers and validates; execution runs. The operator’s behavior, the code that actually transforms records, lives on the exec side, separate from the dispatch plumbing (for the Transform operator, evaluate_single_transform).

Once you add variants 1 and 2, the two exhaustive matches break the build and point you at sites 3 and 4. That’s the closed-enum payoff doing your change-management, the same guarantee that makes a dyn Operator design worse here (a forgotten case there would be a silent runtime no-op, not a compile error).

Add an operator: worked → completion → faded

The real four-place change spans two crates and a lot of supporting types. The shape of it, though, fits in one file: two node enums, a lowering match, a dispatch match. You’ll work that miniature three times: first reading a complete one, then closing one gap, then adding a whole operator end to end. Each rung adds exactly one site of the change.

Worked: the whole shape, compiling

Here is the miniature with all four pieces in place and a Route operator already wired end to end. Run it and watch the lowering-then-dispatch path produce one line per node.

rust // editable

reset ▶ run

// Site 1 — authored side (config). One variant per YAML `type:`.
enum ConfigNode { Transform, Route }

// Site 2 — lowered side (executor vocabulary). A variant per node kind.
enum PlanNode { Transform, Route }

// Site 3 — lowering: config -> plan. Wildcard-free: one arm per ConfigNode.
fn lower(c: &ConfigNode) -> PlanNode {
  match c {
      ConfigNode::Transform => PlanNode::Transform,
      ConfigNode::Route => PlanNode::Route,
  }
}

// Site 4 — dispatch: plan -> operator body. Wildcard-free: one arm per PlanNode.
fn dispatch(p: &PlanNode) -> &'static str {
  match p {
      PlanNode::Transform => "evaluate the CXL program",
      PlanNode::Route => "send the record down a branch",
  }
}

fn main() {
  // Stand in for authored YAML nodes: lower each, then dispatch the result.
  for c in [ConfigNode::Transform, ConfigNode::Route] {
      let p = lower(&c);          // site 3 runs at compile-step time
      println!("{}", dispatch(&p)); // site 4 runs at run time
  }
}

// Site 1 — authored side (config). One variant per YAML `type:`. enum ConfigNode { Transform, Route } // Site 2 — lowered side (executor vocabulary). A variant per node kind. enum PlanNode { Transform, Route } // Site 3 — lowering: config -> plan. Wildcard-free: one arm per ConfigNode. fn lower(c: &ConfigNode) -> PlanNode { match c { ConfigNode::Transform => PlanNode::Transform, ConfigNode::Route => PlanNode::Route, } } // Site 4 — dispatch: plan -> operator body. Wildcard-free: one arm per PlanNode. fn dispatch(p: &PlanNode) -> &'static str { match p { PlanNode::Transform => "evaluate the CXL program", PlanNode::Route => "send the record down a branch", } } fn main() { // Stand in for authored YAML nodes: lower each, then dispatch the result. for c in [ConfigNode::Transform, ConfigNode::Route] { let p = lower(&c); // site 3 runs at compile-step time println!("{}", dispatch(&p)); // site 4 runs at run time } }

Notice the two matches have no _ => arm. That is the whole safety property in miniature: the set of node kinds is closed, so the compiler can check you handled every one.

Completion: add the lowering arm (site 3)

A Dedup operator has been started: the variant is added to both enums (sites 1 and 2), and dispatch already has its arm (site 4). One site is left undone, the lowering arm (site 3). cargo check is reporting a non-exhaustive match at lower. Add the one arm that closes it.

rust // editable

reset ▶ run

enum ConfigNode { Transform, Route, Dedup }
enum PlanNode { Transform, Route, Dedup }

// Site 3 — lowering. This match is non-exhaustive: Dedup is missing.
fn lower(c: &ConfigNode) -> PlanNode {
  match c {
      ConfigNode::Transform => PlanNode::Transform,
      ConfigNode::Route => PlanNode::Route,
      // TODO: add the Dedup arm so this match is exhaustive again.
  }
}

// Site 4 — dispatch already handles Dedup.
fn dispatch(p: &PlanNode) -> &'static str {
  match p {
      PlanNode::Transform => "evaluate the CXL program",
      PlanNode::Route => "send the record down a branch",
      PlanNode::Dedup => "drop a record already seen",
  }
}

fn main() {
  for c in [ConfigNode::Transform, ConfigNode::Route, ConfigNode::Dedup] {
      println!("{}", dispatch(&lower(&c)));
  }
}

enum ConfigNode { Transform, Route, Dedup } enum PlanNode { Transform, Route, Dedup } // Site 3 — lowering. This match is non-exhaustive: Dedup is missing. fn lower(c: &ConfigNode) -> PlanNode { match c { ConfigNode::Transform => PlanNode::Transform, ConfigNode::Route => PlanNode::Route, // TODO: add the Dedup arm so this match is exhaustive again. } } // Site 4 — dispatch already handles Dedup. fn dispatch(p: &PlanNode) -> &'static str { match p { PlanNode::Transform => "evaluate the CXL program", PlanNode::Route => "send the record down a branch", PlanNode::Dedup => "drop a record already seen", } } fn main() { for c in [ConfigNode::Transform, ConfigNode::Route, ConfigNode::Dedup] { println!("{}", dispatch(&lower(&c))); } }

💡 Hint 1

Lowering maps each authored kind to its executor kind. Dedup lowers to Dedup; the names happen to match here, the way Route does (unlike Aggregate→Aggregation).

Show solution

ConfigNode::Dedup => PlanNode::Dedup,

That one arm makes the match exhaustive, and the build goes green. In the real engine this arm is lower_node_to_plan_node’s, and it carries the lowered payload (resolved fields, spans) rather than a bare variant. The move, though, is identical: config kind in, plan kind out, no catch-all to hide a missing case.

Faded: add a whole operator (all four sites)

Now the unaided version. Add a brand-new Filter operator end to end. Both matches start exhaustive over { Transform, Route }, so the moment you add Filter to the enums both will stop compiling: exactly the two-error to-do list you predicted at the top. Add the variant to each enum, then the arm to each match.

rust // editable

reset ▶ run

// TODO site 1: add a `Filter` variant here.
enum ConfigNode { Transform, Route }

// TODO site 2: add a `Filter` variant here too.
enum PlanNode { Transform, Route }

// TODO site 3: add the Filter lowering arm.
fn lower(c: &ConfigNode) -> PlanNode {
  match c {
      ConfigNode::Transform => PlanNode::Transform,
      ConfigNode::Route => PlanNode::Route,
  }
}

// TODO site 4: add the Filter dispatch arm -> the operator body.
fn dispatch(p: &PlanNode) -> &'static str {
  match p {
      PlanNode::Transform => "evaluate the CXL program",
      PlanNode::Route => "send the record down a branch",
  }
}

fn main() {
  for c in [ConfigNode::Transform, ConfigNode::Route, ConfigNode::Filter] {
      println!("{}", dispatch(&lower(&c)));
  }
}

// TODO site 1: add a `Filter` variant here. enum ConfigNode { Transform, Route } // TODO site 2: add a `Filter` variant here too. enum PlanNode { Transform, Route } // TODO site 3: add the Filter lowering arm. fn lower(c: &ConfigNode) -> PlanNode { match c { ConfigNode::Transform => PlanNode::Transform, ConfigNode::Route => PlanNode::Route, } } // TODO site 4: add the Filter dispatch arm -> the operator body. fn dispatch(p: &PlanNode) -> &'static str { match p { PlanNode::Transform => "evaluate the CXL program", PlanNode::Route => "send the record down a branch", } } fn main() { for c in [ConfigNode::Transform, ConfigNode::Route, ConfigNode::Filter] { println!("{}", dispatch(&lower(&c))); } }

Show solution

enum ConfigNode { Transform, Route, Filter }   // site 1
enum PlanNode   { Transform, Route, Filter }   // site 2

fn lower(c: &ConfigNode) -> PlanNode {
    match c {
        ConfigNode::Transform => PlanNode::Transform,
        ConfigNode::Route => PlanNode::Route,
        ConfigNode::Filter => PlanNode::Filter,           // site 3
    }
}

fn dispatch(p: &PlanNode) -> &'static str {
    match p {
        PlanNode::Transform => "evaluate the CXL program",
        PlanNode::Route => "send the record down a branch",
        PlanNode::Filter => "keep only records that pass the predicate", // site 4
    }
}

Four edits, two of them compiler-mandatory (the two match arms). That is the full change-management shape of adding an operator: the closed enums turn “did I wire it everywhere?” into a build error instead of a runtime surprise. In the real engine sites 1–3 land in clinker-plan and site 4 in clinker-exec, and the dispatch arm calls out to the operator’s actual record logic (evaluate_single_transform for Transform) rather than returning a string. The four sites and the two forced arms, though, are exactly these.

Prove it end to end

clinker-exec ·integration_tests.rs ·test_end_to_end_csv_transform test @19acdcb4

#[test]
fn test_end_to_end_csv_transform() {
    // Inline YAML: source -> transform -> transform -> output, CSV in.
    let (counters, dlq, output) = run_pipeline(yaml, csv).unwrap();
    assert_eq!(counters.ok_count, 3);
    assert_eq!(counters.dlq_count, 0);
    // ...assert on the transformed output bytes
}

A new operator’s verify step is a test exactly like this: author a small YAML pipeline that uses your type:, feed input, and assert on the output bytes, testing to the boundary (testing strategy).

// quick check

After adding a PipelineNode::Dedup and a PlanNode::Dedup variant, you run cargo check. What does the compiler report, and why is that the design working as intended?

• Nothing; the new operator runs as a no-op until you write its logic
• Two non-exhaustive-match errors, at lower_node_to_plan_node and at dispatch_plan_node, because both matches are wildcard-free, so they enumerate exactly the arms you still must add
• A single error in the YAML parser about an unknown type

Clear on the four-place change? Skip ahead →

Retrieval checkpoint

✓ Checkpoint — add a transform/operator

💡 Hint 1

Read the match in lower_node_to_plan_node and the match in dispatch_plan_node side by side. Neither has a _ => arm, and each maps one node kind to the next stage. Those are sites 3 and 4.

💡 Hint 2

Note the name change at the Aggregate/Aggregation arm, and that PlanNode::Sort has no PipelineNode twin: sort is injected by the planner, not authored. Build your exercise operator from a YAML-authored kind so you touch all four sites.

What the change establishes

An operator is represented twice across the plan/runtime boundary: PipelineNode (the authored YAML kind, clinker-plan/config) and PlanNode (the executor’s closed enum, clinker-plan/plan). lower_node_to_plan_node maps the first to the second during compile (building the DiGraph<PlanNode> that becomes the ExecutionPlanDag), and dispatch_plan_node maps PlanNode to the operator body at run time. Adding an operator edits four sites (both enum variants plus both match arms), and the two wildcard-free matches make sites 3 and 4 compile-mandatory. The operator’s record logic lives on the clinker-exec side, separate from dispatch.

// quick check

Why does the engine keep two node enums (PipelineNode and PlanNode) connected by a lowering match, instead of one shared enum?

• To make the code longer
• They sit on opposite sides of the plan/runtime boundary: PipelineNode is the authored, span-carrying config shape; PlanNode is the resolved, enriched executor shape (with planner-synthesized variants that were never authored)
• Because serde cannot deserialize the executor enum

// quick check

Put the path one authored node travels in order, from YAML to the operator's record logic.

• YAML → dispatch_plan_node → PlanNode → lower_node_to_plan_node → PipelineNode → operator body
• YAML → PipelineNode → lower_node_to_plan_node → PlanNode → dispatch_plan_node → operator body
• YAML → PlanNode → PipelineNode → operator body (lowering and dispatch run in either order)

// quick check

Adding an operator touches four sites. Which two does the compiler force you to complete (build fails until you do), and which two do you author by choice?

• Forced: the two enum variants. By choice: the two match arms.
• Forced: the two match arms (the lowering arm and the dispatch arm, both in wildcard-free matches). By choice: the two enum variants, which you write first.
• All four are compiler-forced; nothing builds until every site is done.

// quick check

Spaced recall from the through-the-DAG lesson: at run time, how does the executor pick the code to run for each node it walks, and what happens if a node kind is left unhandled?

• It looks the handler up by name in a runtime registry; an unhandled kind is a no-op
• A single match over the closed PlanNode enum (one arm per kind); an unhandled kind is a compile error, because the match has no _ => catch-all
• Each node stores a function pointer it calls; an unhandled kind panics at run time

// quick check

A teammate says: 'PlanNode and PipelineNode are really the same enum with two names; every PlanNode variant came from some YAML type:.' Which statement corrects this MISCONCEPTION?

• Correct as stated: the two enums are always one-to-one, just renamed across the boundary
• PlanNode is a SUPERSET: it has planner-synthesized variants (e.g. Sort, CorrelationCommit) that are injected during compilation and were never authored in YAML, so they have no PipelineNode twin
• It's backwards: PipelineNode is the superset, because YAML can express things the executor can't run

You should be able to:

• You can trace YAML → PipelineNode → (lower) → PlanNode → (dispatch) → operator body
• You can name the four edit sites and which two the compiler forces
• You can explain why PlanNode is a superset of PipelineNode (planner-synthesized nodes)

Verify in the checkout:

grep -n 'pub enum PipelineNode' crates/clinker-plan/src/config/pipeline_node.rs
grep -n 'pub enum PlanNode' crates/clinker-plan/src/plan/execution/mod.rs
grep -n 'fn lower_node_to_plan_node' crates/clinker-plan/src/config/pipeline.rs
grep -n 'fn dispatch_plan_node' crates/clinker-exec/src/executor/dispatch.rs

Connections

You can now extend all three of the engine’s extension points: expressions, formats, and operators. The last two lessons turn from making a change to landing one: the review gauntlet, then planning a change that respects the engine’s boundaries.

Prerequisites

• Add a reader/writer/format

→

Leads to

• Review & the gauntlet

Go deeper on the Rust (optional, one-directional; the same machinery taught from first principles in The Rust Book, with no engine framing):

• Want enum variants from first principles, before they're node kinds? Read the docs →
• Want match exhaustiveness drilled on a small type, the property that forces sites 3 and 4? Read the docs →

Glossary terms used: operator / node.

Mark this lesson complete