Clinker

Clinker is a pure-Rust, bounded-memory batch DAG executor for CSV, JSON, XML, and fixed-width data. It reads finite inputs, drives them through a directed acyclic graph of transformation nodes one record at a time, and exits when the inputs are drained. It ships as a single static binary with no interpreter, no runtime, and no install dependencies.

Pipelines are declared in YAML. Data transformation logic is written in CXL, a custom expression language purpose-built for ETL. Together they replace legacy tools like Informatica, SSIS, Talend, and NiFi with something deterministic, lightweight, and easy to reason about.

What Clinker is, plainly

A finite batch executor with per-record streaming evaluation, not a long-running stream processor. A pipeline run is a job: Sources read until EOF, the DAG drains, the process exits. Within a run, stateless operators (Transform, Route, most Combine probe-side work, Sink) evaluate records one at a time without accumulating per-record state. Every stage is charged against the configured RSS budget. Fused Source → Transform → Sink paths run streaming with no per-stage materialization; non-fused boundaries (Route fan-out, Merge fan-in, Composition bodies, diamond DAGs) materialize records into per-stage buffers that charge against the same envelope. The engine spills buffers to disk at 80% of the limit and fails fast with E310 MemoryBudgetExceeded at the hard limit, naming the offending producer. Blocking operators (Aggregate, sort, grace-hash Combine) accumulate state inside that same budget and spill to disk when soft and hard memory thresholds trip, rather than OOM-killing the process.

If you have used Flink, Kafka Streams, or Beam in unbounded mode: Clinker is not that. There are no watermarks against wall-clock time, no infinite-source semantics, no exactly-once delivery across restarts. The closest prior art is Pentaho Kettle / Apache Hop, Embulk, Singer, Benthos in batch mode, and Vector running file-to-file – finite ETL jobs with per-record evaluation and a hard memory ceiling.

Three pillars of what Clinker is:

1. Finite inputs. Files (CSV, JSON, XML, fixed-width, EDIFACT, X12, HL7 v2, SWIFT MT) are the canonical shape. Finite-cursor network sources (paginated REST APIs with hard page/record caps) fit the same model – they exhaust their cursor and EOF. Unbounded sources (Kafka topics, Kinesis streams, Server-Sent Events, webhooks, tail -f-style file followers) are out of scope and will remain so.
2. Finite jobs. A pipeline run begins when you invoke clinker run, drains the DAG, and exits with a status code. No long-running daemon, no service surface, no infinite event loop.
3. Single process. One clinker binary invocation is one operating- system process. Parallelism happens inside the process via threads (std::thread, Rayon). Clinker does not spawn worker processes, does not coordinate a cluster, and does not shuffle data between machines. Scale by giving the host more cores, more RAM, and more disk – the DuckDB / Polars / Kettle model. If a single host genuinely can’t fit the work, partition the input by file or by key and run multiple clinker invocations from a shell script; that’s a five-line bash script, not an architectural addition.

Why Clinker?

Single binary, zero dependencies. Download it, run it. No JVM, no Python, no package manager. Runs on Linux, macOS, and Windows out of the box — CI builds and tests on all three, and the spill, staging, and RSS-sampling layers have platform-specific paths so behavior is consistent across them.

Good neighbor on busy servers. Clinker enforces a strict memory ceiling (default 512 MB) so it can run alongside JVM applications, databases, and other services without competing for RAM. Aggregation spills to disk when memory pressure rises.

Reproducible output. Given the same input and pipeline, Clinker produces byte-identical output across runs. No nondeterminism from thread scheduling, hash randomization, or floating-point reordering.

Operability-first design. Per-stage metrics, dead-letter queues for error records, explain plans for understanding execution, and structured exit codes for scripting. Built for production from day one.

Two binaries:

A taste of Clinker

Here is a complete pipeline that reads a customer CSV, filters to active customers, classifies them into tiers, and writes the result:

pipeline:
  name: customer_etl

nodes:
  - type: source
    name: customers
    config:
      name: customers
      type: csv
      path: "./data/customers.csv"
      schema:
        - { name: customer_id, type: int }
        - { name: first_name, type: string }
        - { name: last_name, type: string }
        - { name: status, type: string }
        - { name: lifetime_value, type: float }

  - type: transform
    name: enrich
    input: customers
    config:
      cxl: |
        filter status == "active"
        emit customer_id = customer_id
        emit full_name = first_name.concat(" ", last_name)
        emit tier = if lifetime_value >= 10000 then "gold" else "standard"

  - type: sink
    name: result
    input: enrich
    config:
      name: result
      type: csv
      path: "./output/enriched_customers.csv"

Run it:

clinker run customer_etl.yaml

That is the entire workflow. No project scaffolding, no configuration files, no compile step. One YAML file, one command.

Next steps

• Installation – download the binary and verify it works
• Your First Pipeline – build and run a pipeline step by step
• Key Concepts – understand the mental model behind Clinker pipelines

Non-Goals

This page lists what Clinker is deliberately not. These are architectural commitments — design surfaces Clinker will not grow into, not just features that haven’t been built yet.

If you arrived here because you were considering Clinker for one of the scenarios below, the answer is “a different tool is the right fit.” Each non-goal is paired with the kind of tool that is the right fit.

Not an unbounded stream processor

Clinker reads sources that have an end. A pipeline run is a finite job: Sources read until EOF, the DAG drains, the process exits.

Out of scope:

• Kafka topics, Kinesis streams, Pub/Sub subscriptions (long-running consumers without a natural end).
• Server-Sent Events, WebSocket subscriptions, webhooks-as-input.
• tail -f-style file followers.
• Watermarking against wall-clock time.
• Exactly-once delivery across process restarts.
• Stateful infinite-stream windowing (tumbling / sliding / session windows over event time without a finite boundary).

Right fit instead: Apache Flink, Kafka Streams, Apache Beam in unbounded mode, Vector with streaming sources, Benthos with streaming inputs, Apache NiFi.

Not a multi-process or distributed engine

One clinker run invocation is one operating-system process. Clinker does not spawn worker processes, does not coordinate a cluster, and does not shuffle data between machines.

Out of scope:

• Worker-process pools on a single machine.
• Multi-machine sharded execution.
• Network shuffle between executors.
• Cluster managers (Kubernetes operators, YARN, Mesos integrations).
• Distributed memory accounting.
• Partial-failure recovery across worker boundaries.

Right fit instead: Apache Spark, Trino / Presto, Apache Flink in cluster mode, Apache Beam on Dataflow, Hadoop MapReduce.

Scaling Clinker: give the host more cores, more RAM, more disk — the DuckDB / Polars / Kettle / Hop model. If a single host genuinely can’t fit the work, partition the input by file or by key and run multiple clinker invocations from a shell script. That’s a five-line script, not an architectural addition.

Not a long-running service

Clinker is a CLI binary, not a server. There is no daemon mode, no HTTP control plane, no JDBC/ODBC listener, no UI server, no scheduled job runner inside Clinker itself.

Out of scope:

• HTTP API exposing pipeline execution.
• Built-in cron / scheduler / orchestrator.
• Persistent connection pool living across pipeline runs.
• A long-lived process accepting new pipeline submissions over a socket.

Right fit instead:

• For scheduling: cron, systemd timers, Airflow, Dagster, Prefect, Temporal.
• For HTTP-fronted ETL: any of the above orchestrators wrapping clinker run invocations.
• For interactive queries against finite data: DuckDB, Polars, or any embedded query engine.

Orchestration by Temporal (and the others above) is supported via a shell-out contract — the orchestrator runs clinker run as a child process and reads its exit code, logs, and metrics. Clinker embeds no Temporal client or worker; that coupling is a decided non-goal (issue #622). See Running Under a Workflow Orchestrator for the exit-code, cancellation, and output-atomicity guarantees that contract depends on.

Not an OLAP / SQL query engine

Clinker is a per-record expression engine with explicit nodes: in a DAG. It does not parse SQL, does not optimize joins via cost-based optimization across the whole pipeline, and does not present a relational table model.

Out of scope:

• SQL parsing (the CXL language is the surface; no SELECT ... FROM is accepted).
• Cost-based join reordering across more than the local Combine node.
• Materialized views or query caching.
• Interactive query latencies under a second.
• ANSI-SQL semantics for NULL, type coercion, or aggregate behavior.

Right fit instead: DuckDB, ClickHouse, DataFusion, Trino, Postgres, or any RDBMS. If you want SQL-driven transformation over files, DuckDB is the closest single-binary alternative to Clinker for the cases where SQL is the right surface.

Not a connector marketplace

Clinker ships with a deliberately small set of source and sink types: CSV, JSON, XML, fixed-width, EDIFACT, X12, HL7 v2, and SWIFT MT files, plus a finite-cursor REST source. Writing to a network endpoint is not supported: a REST Output sink (issue #224) and finite-cursor SQL sources and sinks (#225, #226) are tracked but unbuilt. There is no plugin registry, no third-party connector store, no SaaS-API catalog.

Out of scope:

• Hundreds of pre-built SaaS integrations (Salesforce, HubSpot, Stripe, etc.).
• A central registry of community-maintained connectors.
• Schema discovery against arbitrary external APIs.
• Change-data-capture (CDC) sources.

Right fit instead: Airbyte, Fivetran, Stitch, Singer with its tap ecosystem, dlt (data load tool).

Not a streaming-CDC engine

Clinker treats each pipeline run as a fresh, finite pass over the input. It does not maintain a persistent log of source changes, does not replicate row-level changes from a database, and does not produce an append-only stream of inserts / updates / deletes.

Out of scope:

• Postgres logical replication subscriptions.
• MySQL binlog tailing.
• Debezium-style CDC stream production.
• Maintaining a target database in continuous sync with a source.

Right fit instead: Debezium, Maxwell, AWS DMS, Striim, Estuary Flow, or vendor-native CDC like Snowflake Streams.

What Clinker is

For the positive framing, see the Introduction and Key Concepts. The short version:

• A pure-Rust, single-binary, bounded-memory batch DAG executor for finite file and finite-cursor inputs.
• Per-record evaluation through a directed acyclic graph of Source, Transform, Aggregate, Route, Merge, Combine, Output, and Composition nodes.
• Pipelines declared in YAML, transformation logic written in CXL (a custom per-record expression language).
• One process, finite job, EOF-then-exit. Disk spill under memory pressure rather than OOM.

Installation

Clinker is a single static binary with no runtime dependencies. Download it, put it on your PATH, and you are ready to go.

Binaries

Clinker ships two binaries:

• clinker – the pipeline executor. This is the main tool you use to validate and run pipelines against data.
• cxl – the CXL expression checker, evaluator, and formatter. Use it during development to test expressions interactively, check types, and format CXL blocks.

Verify installation

After placing the binaries on your PATH, confirm they work:

clinker --version

clinker 0.1.0

cxl --version

cxl 0.1.0

Both commands should print a version string and exit. If you see command not found, check that the directory containing the binaries is in your PATH.

Building from source

Clinker requires Rust 1.91+ (edition 2024). If you have a Rust toolchain installed, build and install both binaries directly from the repository:

# Clone the repository
git clone https://github.com/rustpunk/clinker.git
cd clinker

# Install the pipeline executor
cargo install --path crates/clinker

# Install the CXL expression tool
cargo install --path crates/cxl-cli

This compiles release-optimized binaries and places them in ~/.cargo/bin/, which is typically already on your PATH.

To verify the build:

cargo test --workspace

This runs the full test suite (approximately 1100 tests) and confirms everything is working correctly on your system.

Rust toolchain

The repository includes a rust-toolchain.toml that pins the exact Rust version. If you use rustup, it will automatically download the correct toolchain when you build.

A working C compiler is not one of the requirements: nothing in the build graph runs one. TLS for the rest source and the OTLP exporter goes through rustls with the graviola provider, which ships as Rust and inline assembly rather than the C and per-architecture assembly that other providers build, and content hashing uses blake3’s pure-Rust SIMD paths. CI builds the workspace, including its test and benchmark targets, with every C-compiler environment variable pointed at a program that fails, so a dependency that starts needing one is caught rather than noticed by whoever first builds without a compiler installed. That job also builds a crate that deliberately compiles C and requires it to fail, so the check is known to still work and not merely to be passing.

Graviola supports x86_64 and aarch64. Those are the architectures Clinker is built and tested on; another one needs a different rustls provider, and the ones available today build C.

Optional capabilities

Three parts of Clinker are Cargo features of the clinker crate. All three are on by default — a downloaded binary, or one built with plain cargo install --path crates/clinker, has every one of them, and nothing in this documentation assumes otherwise.

They exist for deployments that want a smaller binary or a narrower dependency graph — a build with no rest and no otlp links no HTTP client and no TLS stack at all:

# File sources and lineage only: no network transport is compiled in.
cargo install --path crates/clinker --no-default-features --features lineage

A build without a capability still parses every construct the full one does. What it will not do is run one silently: a pipeline that declares a rest source, a clinker.toml that sets observability.otlp.endpoint, or a --lineage flag is refused at validation — before any source is opened or any output written — with a diagnostic that names what was asked for and says the capability is not in this binary.

Turning otlp off also stops telemetry being recorded: the fixed arena is reserved for an exporter to drain, so with no exporter there is nothing to reserve it for, and the --machine ndjson-v1 terminal then carries no observability field at all.

Verify a Release

Verify both the archive checksum and its build attestation before running a downloaded Clinker binary. A checksum detects changed bytes; the attestation also binds those bytes to this repository and its release workflow.

Download one platform archive

Replace vX.Y.Z and the archive name with the release you intend to install:

gh release download vX.Y.Z \
  --repo rustpunk/clinker \
  --pattern 'clinker-vX.Y.Z-x86_64-unknown-linux-gnu.tar.gz*'

Each native archive has a sibling .sha256 file. The release also contains a SHA256SUMS inventory covering all supported archives.

Check the SHA-256 digest

On Linux:

sha256sum --check clinker-vX.Y.Z-x86_64-unknown-linux-gnu.tar.gz.sha256

On macOS:

shasum -a 256 -c clinker-vX.Y.Z-aarch64-apple-darwin.tar.gz.sha256

On Windows PowerShell:

$archive = "clinker-vX.Y.Z-x86_64-pc-windows-msvc.zip"
$expected = (Get-Content "$archive.sha256").Split()[0].ToLowerInvariant()
$actual = (Get-FileHash -Algorithm SHA256 $archive).Hash.ToLowerInvariant()
if ($actual -ne $expected) { throw "Clinker archive checksum mismatch" }

Stop if the checksum command fails. Do not extract or run the archive.

Verify build provenance

With a current GitHub CLI and network access:

gh attestation verify \
  clinker-vX.Y.Z-x86_64-unknown-linux-gnu.tar.gz \
  --repo rustpunk/clinker

The verification must identify rustpunk/clinker as the source repository and the release archive itself as the attested subject. An attestation proves where and how the bytes were built; it is not a claim that the program is free of security defects.

If either checksum or provenance verification fails, keep the archive quarantined and report the release tag, archive name, and failing command.

Your First Pipeline

This walkthrough builds a pipeline from scratch, runs it, and explores the tools Clinker provides for validating and understanding pipelines before they touch real data.

1. Create sample data

Save the following as employees.csv:

id,name,department,salary
1,Alice Chen,Engineering,95000
2,Bob Martinez,Marketing,62000
3,Carol Johnson,Engineering,88000
4,Dave Williams,Sales,71000

2. Write the pipeline

Save the following as my_first_pipeline.yaml:

pipeline:
  name: salary_report

nodes:
  - type: source
    name: employees
    config:
      name: employees
      type: csv
      path: "./employees.csv"
      schema:
        - { name: id, type: int }
        - { name: name, type: string }
        - { name: department, type: string }
        - { name: salary, type: int }

  - type: transform
    name: classify
    input: employees
    config:
      cxl: |
        emit id = id
        emit name = name
        emit department = department
        emit salary = salary
        emit level = if salary >= 90000 then "senior" else "junior"

  - type: sink
    name: report
    input: classify
    config:
      name: report
      type: csv
      path: "./salary_report.csv"

This pipeline has three nodes:

1. employees (source) – reads the CSV file and declares the schema.
2. classify (transform) – passes all fields through and adds a level field based on salary.
3. report (sink) – writes the result to a new CSV file.

The input: field on each consumer node wires the DAG together. Data flows from employees through classify to report.

3. Validate before running

Before processing any data, check that the pipeline is well-formed:

clinker run my_first_pipeline.yaml --dry-run

Dry-run parses the YAML, resolves the DAG, and type-checks all CXL expressions against the declared schemas. If there are errors – a typo in a field name, a type mismatch, a missing input: reference – Clinker reports them with source-location diagnostics and stops. No data is read.

4. Preview records

To see what the output will look like without writing files, preview a few records:

clinker run my_first_pipeline.yaml --dry-run -n 2

This reads at most 2 records from this source, runs them through the pipeline, and writes the preview to stdout without opening salary_report.csv. For a pipeline with several Sources, the limit applies separately to each Source; output counts can differ after filters, joins, or aggregates. Use --dry-run-output preview.csv to select an explicit preview destination.

At revision 3b343a4e, this example’s bounded preview can fail with an internal node-buffer cleanup error. Its ordinary run produces the output below. See the preview limitation.

5. Understand the execution plan

To see how Clinker will execute the pipeline:

clinker run my_first_pipeline.yaml --explain

The explain plan shows the DAG topology, the order nodes will execute, per-node parallelism strategy, and schema propagation through the pipeline. This is valuable for understanding complex pipelines with routes, merges, and aggregations.

6. Run it

clinker run my_first_pipeline.yaml

Clinker reads employees.csv, applies the transform, and writes salary_report.csv. The output:

id,name,department,salary,level
1,Alice Chen,Engineering,95000,senior
2,Bob Martinez,Marketing,62000,junior
3,Carol Johnson,Engineering,88000,junior
4,Dave Williams,Sales,71000,junior

Alice’s salary of 95,000 meets the threshold, so she is classified as senior. Everyone else is junior.

What just happened

The pipeline executed as a streaming process:

1. The source node read employees.csv one record at a time.
2. Each record flowed through the classify transform, which evaluated the CXL block to produce the output fields.
3. The Sink node wrote each transformed record to salary_report.csv.

At no point was the entire dataset loaded into memory. This is how Clinker processes files of any size under its memory ceiling.

Next steps

• Key Concepts – understand the building blocks of Clinker pipelines
• Pipeline YAML Structure – full reference for pipeline configuration
• CXL Overview – learn the expression language in depth

Key Concepts

This page covers the mental model behind Clinker pipelines. If you have experience with other ETL tools, most of this will feel familiar – but pay attention to where Clinker diverges, especially around Clinker Expression Language (CXL), per-record evaluation, and the memory budget.

Batch jobs, not unbounded streams

A Clinker run is a finite batch job. Source nodes read their files until EOF, the DAG drains, and the process exits. There are no watermarks against wall-clock time, no infinite-source semantics, no exactly-once delivery across restarts. If you have used Flink, Kafka Streams, or Beam in unbounded mode: Clinker is not that.

The word “streaming” in Clinker’s documentation always refers to per-record evaluation within a single batch run – records flow through the graph one at a time rather than being materialized as a whole table – not to long-running stream-processor semantics. Internal identifiers in the codebase (function names like streaming_output_task, config fields like strategy: streaming, error messages, log lines) use the word in the same row-by-row sense; if you see it in a stack trace, it is not Flink leaking through.

Finite inputs only

Clinker reads sources that have an end. Files are the canonical shape, and finite-cursor network sources (paginated REST APIs with hard page/record caps) fit the same model – they exhaust their cursor and EOF. Unbounded sources (Kafka, Kinesis, Server-Sent Events, webhooks, tail -f-style file followers) are explicitly out of scope and will remain so.

Single process, ever

One clinker run invocation is one OS process. Parallelism happens inside that process via threads. Clinker does not spawn worker processes, does not coordinate a cluster, and does not shuffle data between machines. Scale by giving the host more cores, more RAM, more disk – the DuckDB / Polars / Kettle model. If a single host genuinely can’t fit the work, partition the input by file or by key and run multiple clinker invocations from a shell script; that’s a five-line script, not an architectural addition to Clinker.

For the full list of what Clinker deliberately does not do, see Non-Goals.

Pipelines are DAGs

A pipeline is a directed acyclic graph of nodes. Data flows from sources, through processing nodes, to outputs. There are no cycles – a node cannot consume its own output, directly or indirectly.

You define the graph with each node kind’s input fields. Most consumers use a single input:, Merge uses an inputs: list, Combine uses a named input: map, and Envelope names its body plus optional header and trailer streams. Clinker resolves these references, validates that the graph is acyclic, and determines execution order automatically.

From YAML to a run

Clinker first parses and validates the pipeline, binds schemas, compiles CXL, and produces a typed CompiledPlan. Execution entry points require that plan, so invalid pipelines are rejected before records are read. The current runtime then recompiles the configuration embedded in the supplied plan and executes the newly validated artifacts. It does not yet execute the stored DAG and other stored artifacts directly.

The locked D-01 through D-11 contract changes that lifecycle: the supplied CompiledPlan becomes authoritative, survives execution, and may be reused sequentially in one process while only a defined runtime envelope may refresh. Phase 5 owns that correction and any versioned persistent plan cache. Direct stored-plan execution and persistent caching are not current capabilities. See Stored-plan execution and cache identity for the current status and owner of each contract.

The nodes: list

Every pipeline has a single flat list of nodes. Each node has a type: discriminator that determines its behavior. The eleven node types are:

You can have as many nodes of each type as your pipeline requires. The only constraint is that the resulting graph must be a valid DAG.

CXL is per-record ETL

CXL is Clinker’s per-record ETL expression language. Each record flows through a CXL block independently. A program maps, filters, or enriches the current record in statement order; it is not a table query or join planner.

The core statements:

• emit name = expr – produce a field in the output record, adding it or replacing the field of the same name. The record’s other fields are carried through unchanged, so there is no need to write emit id = id; a Sink can narrow what it writes (see Sink Nodes).
• let name = expr – bind a local variable for use in later expressions. Local variables do not appear in the output.
• filter condition – discard the record if the condition is false. A filtered record produces no output and is not counted as an error.
• distinct / distinct by field – deduplicate records. distinct deduplicates on all output fields; distinct by field deduplicates on a specific field.

CXL uses and, or, and not for boolean logic – not && or ||. String concatenation uses +. Conditional expressions use if ... then ... else ... syntax.

System namespaces use a $ prefix: $pipeline.*, $source.*, $record.*, $window.*, and $vars.*. These provide access to pipeline and source context, per-record scoped state, window function state, and static configuration respectively.

Per-record evaluation and the memory budget

Within a run, stateless operators evaluate records one at a time: a CXL block sees exactly one record, with no table-level context and no per-record state carried across records. Clinker does not load an entire file into memory before processing it. This is what “streaming” means in Clinker – row-by-row evaluation inside a finite batch job, not Flink-style unbounded stream processing.

“One at a time” describes the evaluation model, not the transport. For efficiency, records move between stages in bounded batches (default 2048 events, tunable per stage or pipeline-wide via batch_size) rather than as individual messages – but each record is still evaluated independently against the CXL block. The batch is a handoff unit, not a window: an operator never needs the whole batch to process any single record. (Blocking operators – Aggregate, sort, grace-hash Combine – are the exception that do accumulate across records; see below.)

Per-record evaluation keeps per-row memory usage bounded for the stateless parts of the graph (Transform, Route, Merge, most Combine probe-side work, Output). Every stage is charged against the configured RSS budget. Fused Source → Transform → Sink paths run streaming, with no per-stage materialization, so a 100 GB CSV passes through with the same footprint as a 100 KB CSV. A stage that hands its output to a single downstream Sink also avoids a charged inter-stage buffer – single-branch Route, non-fused Merge, streaming Aggregate, and the Combine probe-side stream their result straight to the writer (see Streaming vs. Blocking Stages). The remaining boundaries – multi-branch Route fan-out, output that forks to several consumers, Composition bodies, diamond DAGs – materialize records into per-stage buffers that charge against the same budget envelope. Every materialized buffer can spill past the soft threshold, including buffers shared by several readers and Route/Cull output-port buffers. Readers run sequentially over the same immutable memory-or-spill backing; each opens one cursor, and the final reader takes the authoritative buffer regardless of declaration or dispatch order. A consumer that needs a full resident vector reserves that materialization first. If the overlap would exceed the hard limit, the engine fails before allocating with a structured E310 MemoryBudgetExceeded diagnostic that names the consumer.

Use clinker run --explain to see which nodes will materialize (buffer: materialized) versus which will stream (buffer: streaming) before runtime – that label is the canonical “which stages charge the budget” signal. See the --explain reference and the memory-tuning page.

Stateful operators must accumulate. Aggregate, sort, and grace-hash Combine cannot emit until they have seen enough input – sums need every addend, a full sort needs the last row, a hash join needs the build side complete. These operators run inside a configured RSS budget (default 512 MB) and degrade gracefully under pressure rather than OOM:

• Aggregate uses hash aggregation by default and spills partitions to disk when soft/hard memory thresholds trip. When the input is already sorted by the group key, the planner picks streaming aggregation, which requires only constant memory.
• Sort spills runs to disk and merges them.
• Combine picks among in-memory hash join, grace hash join (spilled), and IEJoin / sort-merge depending on predicates and memory pressure. A pure-range Combine (band join with no equality key) runs the block-band IEJoin, which external-sorts each side and spills a matched-output sort to disk so both its input and its result stay inside the budget.

The memory ceiling is a first-class promise. Clinker is designed to share a server with JVM applications, databases, and other services without competing for RAM.

Input wiring

Consumer nodes reference their upstream via the input: field:

- type: transform
  name: enrich
  input: customers    # reads from the node named "customers"

Route nodes produce named output ports. Downstream nodes reference a specific port using dot notation:

- type: route
  name: split_by_region
  input: customers
  config:
    routes:
      us: region == "US"
      eu: region == "EU"
    default: other

- type: sink
  name: us_output
  input: split_by_region.us    # reads from the "us" port

Merge nodes accept multiple inputs using inputs: (plural):

- type: merge
  name: combined
  inputs:
    - us_transform
    - eu_transform

Combine, Envelope, and Composition have their own multi-input shapes. See Pipeline YAML Structure for the complete wiring table.

Schema declaration

Source nodes require an explicit schema: that declares every column’s name and type:

config:
  schema:
    - { name: customer_id, type: int }
    - { name: email, type: string }
    - { name: balance, type: float }
    - { name: created_at, type: date }

Clinker uses these declarations to type-check CXL expressions at compile time, before any data is read. If a CXL block references a field that does not exist in the upstream schema, or applies an operation to an incompatible type, the error is caught during validation – not at row 5 million of a production run.

Supported types include int, float, string, bool, date, and datetime.

Error handling

A pipeline picks one error handling strategy, in the top-level error_handling: block:

When using continue, Clinker writes rejected records to a DLQ file alongside the output. Each DLQ entry includes the original record, the error category, the error message, and the node that rejected it. This makes diagnosing production issues straightforward: check the DLQ, fix the data or the pipeline, and rerun. A run that dead-letters at least one record exits with code 2 rather than 0, so a scheduler can tell a clean run from a partial one.

See Error Handling & DLQ for the DLQ columns, the error categories, and the per-source options.

Interactive Guides

Some parts of Clinker are easier to understand by trying them than by reading about them. Each guide below is a single page that runs in your browser, on a desktop or a phone. You change a setting or tap a record, and the page shows what the engine does with it. The guides don’t run a pipeline; each one reproduces the rules described on the reference page it links back to.

Where does a null go?

How CXL works out an expression when a field is empty: and, or and not with null, why null == null is true, and why a filter drops a record whose condition comes out null. Reference: Null Handling.

Correlation keys

How one failing line takes the rest of its order to the DLQ, and how an Aggregate’s group_by decides between rejecting a whole group and recomputing totals without the failed line. Reference: Correlation Keys.

Document context

Where $doc.* values come from, how each file becomes its own document with its own Aggregate roll-up, and what dlq_granularity: document rejects. Reference: Document Envelope Context.

How many documents?

What an Envelope node’s preserve and concat do to the documents a Sink writes, how a synthesized footer differs between them, how an Aggregate after the Envelope rolls up, and which pipeline shapes E347 and E355 reject. Reference: Envelope Nodes.

Which files get read?

How a Source turns glob:, regex:, paths: and its filters into the list of files it reads, and in what order. Reference: Source Nodes → Choosing files.

Is my data still sorted?

Which stages keep a Source’s declared sort_order and which drop it, and when an Aggregate can stream instead of holding every group. Reference: Aggregate Nodes and Source Nodes.

Route and Merge

Where each record goes when a Route’s conditions are true, not true, or fail, in exclusive and inclusive mode, and how a Merge rejoins the branches. Reference: Route Nodes and Merge Nodes.

Combine playground

Which build rows where: matches for each driver row, and what match:, on_miss: and drive: do with them, including range joins. Reference: Combine Nodes.

Window functions

Which rows of a partition each $window.* function reads, and why $window.sum is a partition total while $window.cumulative_sum is the running total. Reference: Window Functions.

Where did my rows go?

What the end-of-run line N total, N ok, N written, N dlq counts, why the numbers often don’t add up, and which rows are in no number at all: filtered and duplicate rows, rows folded into Aggregate groups, Combine misses and Route branches nothing reads. Reference: Metrics & Monitoring.

Streaming vs. blocking

Which stages stream and which hold their output in a buffer, for several pipeline shapes, with the --explain lines each one produces. Reference: Streaming vs. Blocking Stages.

For engine developers

The Clinker Engine Internals book has three more:

• a memory system explainer, linked from its Memory Arbitration & Scheduling chapter. It covers the memory budget, back-pressure, spilling to disk and the scheduler, with a simulator.
• a range-join explainer, linked from its Combine Internals chapter. It runs the block-band IEJoin on small inputs: sorting and slicing into blocks, pruning block pairs, the kernel step by step, and the nested-loop fallback.
• a retraction-loop explainer, linked from its Retraction Protocol chapter. It replays the commit loop that corrects an Aggregate whose group_by leaves out a correlation-key field, on three of the engine’s test pipelines.

Pipeline YAML Structure

A Clinker pipeline is a single YAML file with three top-level sections: pipeline (metadata), nodes (the processing graph), and optionally error_handling.

Top-level shape

pipeline:
  name: my_pipeline            # Required — pipeline identifier
  memory:                      # Optional — see ops/memory.md
    limit: "256M"              # Optional (K/M/G suffixes), default 512M
    backpressure: pause        # Optional, default `pause`
  vars:                        # Optional typed static configuration
    threshold: { type: int, default: 500 }
    label: { type: string, default: "Monthly Report" }
  date_formats: ["%Y-%m-%d"]   # Optional — custom date parsing formats
  rules_path: "./rules/"       # Optional — CXL module search path
  concurrency:                 # Optional
    threads: 4
    chunk_size: 1000
  metrics:                     # Optional
    spool_dir: "./metrics/"

nodes:                         # Required — flat list of pipeline nodes
  - type: source
    name: raw_data
    config:
      name: raw_data
      type: csv
      path: "./data/input.csv"
      schema:
        - { name: id, type: int }
        - { name: value, type: string }

  - type: transform
    name: clean
    input: raw_data
    config:
      cxl: |
        emit id = id
        emit value = value.trim()

  - type: sink
    name: result
    input: clean
    config:
      name: result
      type: csv
      path: "./output/result.csv"

error_handling:                # Optional
  strategy: fail_fast

Pipeline metadata

The pipeline: block carries global settings that apply to the entire run.

These three names are admitted only far enough to produce precise, spanned diagnostics. Empty strings/maps and false are still explicit values and are rejected before execution; omission is the only accepted form.

Reserved metadata contract

The current status and locked owner are explicit:

For provenance sidecars that work today, set write_meta: true on an Output node. D-24 keeps that spelling. See Approved exceptions and rejected placeholders.

The nodes list

Every pipeline has a flat nodes: list. Each entry is a node with a type: discriminator that determines its kind:

Node naming

Every node must have a name: field. Names must be unique within the pipeline and must not contain dots – the dot addresses something other than a node: a route branch (split.high, see below) and a node inside a composition call site (enrich.ref). The rule covers every node kind, including sources, outputs, and composition call sites, and it covers nodes declared inside a .comp.yaml body. Names are used for wiring, logging, and diagnostics.

A dotted name is refused at plan time with E010, which names the node and the name to use instead:

node name "enrich.ref" is invalid: '.' is reserved for branch references and
composition call-site paths; rename the node to "enrich_ref" (use underscores
or hyphens) and update every reference to it

Wiring by node kind

Input fields live at the node’s top level, alongside name: and type:. Their shape is specific to the node kind:

Single upstream – used by ordinary one-input consumers:

- type: transform
  name: clean
  input: raw_data       # References the source node named "raw_data"
  config: ...

Port syntax – for consuming a specific branch from a route node, use node.port:

- type: sink
  name: high_value_out
  input: split.high     # Consumes the "high" branch of route node "split"
  config: ...

Multiple upstreams – merge nodes use inputs: (plural) instead of input::

- type: merge
  name: combined
  inputs:
    - east_processed
    - west_processed
  config: {}

Qualified inputs – Combine uses singular input: with a map whose keys become CXL qualifiers:

- type: combine
  name: enriched
  input:
    orders: clean_orders
    products: product_catalog
  config:
    where: "orders.product_id == products.product_id"
    match: first
    on_miss: null_fields
    cxl: |
      emit order_id = orders.order_id
      emit product_name = products.name
    propagate_ck: driver

Envelope and Composition have additional port semantics. See Envelope Nodes and Compositions before wiring those node kinds.

Source nodes have no input field. They are entry points – adding an input: field to a source is a parse error.

Using the wrong field or value shape for a node kind is caught at parse time by strict deserialization.

Optional fields on all nodes

Every node type supports these optional fields:

• description: – human-readable text for documentation. Ignored by the engine.
• _notes: – arbitrary metadata (JSON object). Ignored by the engine and available to external tooling.

- type: transform
  name: enrich
  description: "Add customer tier based on lifetime value"
  _notes:
    color: "#4a9eff"
    position: { x: 300, y: 200 }
  input: customers
  config:
    cxl: |
      emit tier = if lifetime_value >= 10000 then "gold" else "standard"

Strict parsing

All config structs use deny_unknown_fields. If you misspell a field name – for example, writing inputt: instead of input: or stratgy: instead of strategy: – the YAML parser rejects it immediately with a diagnostic pointing to the typo. This catches configuration errors before any data processing begins.

Environment variable: CLINKER_ENV

The CLINKER_ENV environment variable can be used for conditional logic outside of pipelines (e.g., selecting channel directories or controlling CLI behavior). It is not directly referenced within pipeline YAML but is available to the channel and workspace systems.

Scoped Variables

Clinker’s scoped-variable system lets a pipeline read and write named values at three lifetimes: the pipeline run, the source, and the record. Each variable is declared on the Transform that writes it via that Transform’s declares: block (type, scope, optional default), written by the same Transform’s CXL with emit $<scope>.<name> = ..., and read inline from any downstream node via the $pipeline.*, $source.*, and $record.* namespaces.

The three scopes

Record-scope variables are the per-record private store: every transform along the row’s path can read them, but they never serialize as output columns unless explicitly re-emitted as a regular column. They are written with emit $record.<key> = ... from a transform that declares them.

Declaring variables

A scoped variable is declared on the Transform that writes it, in that Transform’s config.declares: list. Each entry is named, scoped, typed, and optionally given a default that satisfies reads firing before the writer has run:

- type: transform
  name: enrich
  input: orders
  config:
    declares:
      - { name: cutoff_date,  scope: pipeline, type: date,   default: "2024-01-01" }
      - { name: ingest_label, scope: source,   type: string, default: "prod" }
      - { name: fuzzy_score,  scope: record,   type: float }
    cxl: |
      emit id = id
      emit $pipeline.cutoff_date = "2024-01-01"
      emit $source.ingest_label = $source.file.file_stem()
      emit $record.fuzzy_score = fuzzy_match(name, $pipeline.canonical_name)

Allowed types: int, float, string, bool, date, date_time.

Each (scope, name) pair must be declared on exactly one Transform — the same pair declared on two Transforms is rejected at config-validation time, ahead of compilation. $pipeline, $source, and $record are flat shared namespaces; declare each name once and read it from every consumer.

The pipeline’s top-level vars: block is a separate, flat registry for static configuration read via $vars.<key> — it does not carry the nested pipeline: / source: / record: scopes:

pipeline:
  name: order_processing
  vars:
    fuzzy_threshold: { type: float, default: 0.85 }   # read as $vars.fuzzy_threshold

Built-in members of each scope ($source.file, $source.name, $source.row, $source.path, $source.count, $source.batch, $source.ingestion_timestamp; $pipeline.start_time, $pipeline.name, $pipeline.execution_id, $pipeline.batch_id, $pipeline.total_count, $pipeline.ok_count, $pipeline.dlq_count, $pipeline.filtered_count, $pipeline.distinct_count) are reserved — declaring a user variable with one of those names is rejected at parse time.

$source.count semantics

$source.count is the per-source record total for the Source that produced the current record. The total isn’t known until the source finishes, so you can’t use it during per-record evaluation: a read on a mid-stream record (in a Transform, Route, Window, or Merge) resolves to Null, while reads after the source has finished (such as a terminal aggregate emit) resolve to the per-source total.

This means a streaming denominator like value / $source.count yields Null on mid-stream records. If you need a running row counter, declare a scope: source variable on a Transform and increment it from that Transform’s CXL instead.

Reading variables

CXL access is identical for declared and built-in keys:

- type: transform
  name: filter_recent
  input: orders
  config:
    cxl: |
      emit id = id
      filter received_at > $pipeline.cutoff_date
      emit batch = $source.batch_id
      emit confidence = $record.fuzzy_score

Reads of undeclared keys are rejected with E203 (CXL name resolution failed) at compile time, with a “did you mean” suggestion that scans the declared registry.

Writing variables

A scoped variable is written by the Transform that declares it: list the variable in the Transform’s declares: block and assign it from the same Transform’s CXL with emit $<scope>.<name> = <expr>. The Transform still processes records normally — declaring and writing a scoped var is additive to its ordinary emit/filter logic.

- type: transform
  name: capture_header
  input: salesforce_in
  config:
    declares:
      - { name: batch_id,        scope: source, type: string }
      - { name: ingestion_label, scope: source, type: string }
    cxl: |
      emit id = id
      emit $source.batch_id = batch
      emit $source.ingestion_label = $source.file.file_stem()

- type: transform
  name: row_score
  input: enrich
  config:
    declares:
      - { name: fuzzy_score, scope: record, type: float }
    cxl: |
      emit id = id
      emit $record.fuzzy_score = fuzzy_match(name, $pipeline.canonical_name)

An emit $<scope>.<name> write to a variable the Transform does not declare is rejected at compile time. Requiring the declares: entry keeps the dependency between writers and readers visible at plan time.

Init phase: pre-runtime population

Set phase: init on a Transform to pre-compute a $pipeline.* or $source.* value from a config-file source before the main run starts:

- type: source
  name: config_src
  config:
    name: config_src
    type: csv
    path: config.csv
    schema:
      - { name: cutoff, type: int }

- type: aggregate
  name: max_agg
  input: config_src
  config:
    group_by: []
    cxl: |
      emit cap = max(cutoff)

- type: transform
  name: precompute_cutoff
  input: max_agg
  config:
    phase: init
    declares:
      - { name: cutoff_date, scope: pipeline, type: int }
    cxl: |
      emit cap = cap
      emit $pipeline.cutoff_date = cap

Init-phase nodes must be terminal — no runtime-phase node may consume from an init-phase Transform. (Init-phase nodes can chain through init-only descendants for compositions.) Use disjoint Sources for init vs runtime when you need both: a Source shared between an init and a runtime branch only feeds the init pass.

Compile-time validation

Scoped variables are checked before the run starts. Every reference and every writer is validated, and every flow from a writer to its readers is checked against the pipeline. Each code below tells you what to fix.

Cross-Transform duplicate declares: (the same (scope, name) declared on two Transforms) is rejected before the run starts. $pipeline, $source, and $record are flat shared namespaces; declare each name once and reference it from every consumer.

Each diagnostic points at the exact place you read or wrote the variable, plus the conflicting writer or parent declaration, so the report lands where you can act on it rather than in some unrelated configuration block.

Post-merge access: qualified $source.<input>.<key>

After a Merge or Combine, the bare $source.<custom> form is ambiguous: each record carries its own source’s value, but the reader’s intent is usually to compare across inputs. E172 rejects the unqualified form and the qualified form is the legal alternative:

- type: transform
  name: read_after_merge
  input: merged
  config:
    cxl: |
      emit id = id
      emit lt = $source.left_input.left_label
      emit rt = $source.right_input.right_label

The <input_name> segment matches the named input on the Combine (its IndexMap key) or the upstream node name on the Merge.

Composition opt-in

A composition body cannot see parent scoped variables by default — the seal is enforced by E173. To pass values across the boundary, the composition declares the schema of parent vars it consumes in its _compose.scoped_vars block:

# read_pipeline_var.comp.yaml
_compose:
  name: read_pipeline_var
  inputs:
    inp:
      schema:
        - { name: id, type: int }
  outputs:
    out: tap
  scoped_vars:
    pipeline:
      cutoff:
        type: int

nodes:
  - type: transform
    name: tap
    input: inp
    config:
      cxl: |
        emit id = id
        emit cutoff_seen = $pipeline.cutoff

The parent must declare cutoff with the matching type; mismatches raise E174.

What scoped variables are not

These are intentional non-features:

• No persistence across runs. State is in-memory only. A pipeline run starts with declaration defaults; the writes don’t survive the process.
• No undeclared writes. A Transform may only write a scoped variable it lists in declares:; an emit $pipeline.x to an undeclared name is a compile error. Requiring the declaration keeps every writer visible at plan time and the writer→reader dependency explicit in the DAG.
• No dynamic var creation. The set of variables is closed at plan time, by design. This bounds memory and makes the validation matrix above tractable.

Channel overrides

A channel can both override a pipeline’s declaration defaults and add new entries across all four registries ($vars.*, $pipeline.*, $source.*, $record.*). Each registry has its own sub-block under vars: on a .channel.yaml, and each entry uses the same { type, default } shape that pipeline-side declarations use:

# Pipeline declarations
pipeline:
  name: orders
  vars:
    fuzzy_threshold: { type: float, default: 0.85 }   # $vars.*
nodes:
  - type: source
    name: orders_src
    config: { name: orders_src, type: csv, path: in.csv,
              schema: [{ name: id, type: int }] }
  - type: transform
    name: enrich
    input: orders_src
    config:
      declares:
        - { name: cutoff_date,  scope: pipeline, type: date,   default: "2024-01-01" }
        - { name: ingest_label, scope: source,   type: string, default: "prod" }
        - { name: tier,         scope: record,   type: string, default: "bronze" }
      cxl: |
        emit id = id

# channel/acme-prod/orders.channel.yaml
channel:
  target: ../../pipelines/orders.yaml
vars:
  static:
    fuzzy_threshold: { type: float, default: 0.95 }
  pipeline:
    cutoff_date: { type: date, default: "2026-01-01" }
  source:
    orders_src:
      ingest_label: { type: string, default: "acme-prod" }
  record:
    tier: { type: string, default: "platinum" }

The overlay lives in the tenant’s folder (channel/acme-prod/) and is applied with --channel acme-prod; the channel.target field is authoritative.

Override semantics (entry name already declared) require the channel’s type to match the declared type — mismatches produce E116. Add semantics (entry name not yet declared) extend the registry with a new declaration. In both cases, a default that does not match the entry’s type also produces E116. $source overrides are keyed by source-node name; an unknown source name produces E118. The reserved-name guard (E117) blocks channels from shadowing system fields like $pipeline.execution_id or $source.path. Channels that target a .comp.yaml may not carry vars: (E109 if they do).

See Channels for the full overlay rules and the channel manifest reference.

Channels

Channels make one pipeline serve many tenants. A single base pipeline is authored once; each tenant (a channel) layers its own configuration, variable defaults, and structural changes on top — without copying or editing the base YAML. The system is built for scale: thousands of per-tenant channels against one pipeline, with strict validation and per-value provenance.

A channel is a tenant. A group is a reusable overlay shared by many channels — selected automatically from a channel’s labels, or invoked by name. Groups and target files can contribute value clobber (config: / vars: / resources:) and an ordered op list (overrides:). A channel-wide manifest is narrower: it may contain only labels plus declared config and variables.

Workspace layout

Channels live in a channel-centric workspace. A clinker.toml at the workspace root declares the layout roots; the rest is folders of YAML:

workspace/
  clinker.toml                       # declares the [channel] and [group] roots
  pipeline/       *.yaml             # base pipelines  (the pipeline-default layer)
  composition/    *.comp.yaml        # reusable sub-pipelines
  schema/         *.schema.yaml      # shared schemas
  group/          *.group.yaml       # group overlays: selector, priority, overrides
  channel/<tenant>/                  # one cataloged channel resource folder
    channel.cfg.yaml                 # required manifest: identity, targets, labels, wide values
    orders.yaml                      # filename is descriptive; channel.target is authoritative

The channel id is the stable logical key in [catalog.channels]. A --channel tenant.globex invocation resolves that catalog entry directly and then selects the target file by its declared logical pipeline id. Neither a folder name, a file basename, nor the current working directory is an identity.

clinker.toml roots

[channel]
root = "channel"      # per-channel folders live under <root>/<channel-id>/
shard = "none"        # enumeration layout: none (default) | first-char | hash

[group]
root = "group"        # *.group.yaml definitions live here

Both tables are optional; omitting them defaults [channel].root to channel, [channel].shard to none, and [group].root to group. shard is an enumeration-ergonomics choice for very large channel trees (it splits the folder fan-out); a channel is always looked up by computed path regardless of shard scheme, so shard never changes resolution semantics.

Typed workspace catalog

The same clinker.toml declares stable logical identities in separate catalog namespaces for rules, schemas, compositions, pipelines, channels, and typed composition resources.

[catalog]
rules_root = "rules"

[catalog.rules]
"shared.dates" = "rules/shared/dates.cxl"

[catalog.schemas]
"shared.dates" = "schema/shared/dates.schema.yaml"

[catalog.compositions]
"etl.normalize" = "composition/normalize.comp.yaml"

[catalog.pipelines]
"daily.orders" = "pipeline/orders.yaml"

[catalog.channels]
"tenant.globex" = "channel/globex"

[catalog.resources.shared_orders]
kind = "file"
path = "data/orders.csv"
access = "read"

Logical identities are kind-scoped. The rule and schema named shared.dates above are distinct typed resources; asking for one kind never substitutes an entry from another kind. A missing identity or a reference through the wrong kind fails planning and names the catalog table that must contain it.

catalog.resources is additionally descriptor-typed. The current file kind accepts only path and access (read, write, or read-write), is admitted under fixed catalog entry and descriptor-byte caps, and contains no credential fields. Resource bindings use the logical key (shared_orders above), never the path.

Every catalog path and rules root is anchored to the selected workspace (from --base-dir or workspace discovery). Parent traversal, an absolute path outside the workspace, and a symlink whose canonical target escapes the workspace are rejected. Duplicate identities within one kind are rejected, as are two catalog identities—even across kinds—that alias the same canonical file. These checks happen before compilation, so neither lexical aliases nor symlink aliases can create a hidden second authority for one file.

For CXL modules, an explicit rule entry is used when present; otherwise the logical identity maps beneath one selected rules root. Root precedence is explicit CLI --rules-path, then pipeline.rules_path, then [catalog].rules_root, then the workspace-relative rules/ default. Selection chooses one root rather than searching several. Planning admits the bounded direct/transitive module closure into the compiled plan, after which execution does not reopen module source files. See Modules and use and the --rules-path reference.

The layer model

Every value and every op is attributed to exactly one layer. Layers apply in a fixed semantic order — never lexical or file order:

pipeline-default  <  group(s) by priority  <  channel-wide  <  channel-per-target

• pipeline-default — the base pipeline’s own configuration.
• group(s) by priority — every group applied to the run, ordered by priority (higher priority applies later and thus wins).
• channel-wide — the channel manifest (channel.cfg.yaml): overlays that apply to every pipeline this channel runs.
• channel-per-target — the per-target overlay file (<target>.channel.yaml): the highest-precedence layer.

Clobber, never deep-merge

A higher layer’s value replaces the lower layer’s value wholesale. There is no deep-merge and no list-append: overriding a list swaps the entire list. To override individual elements, model them as a keyed map (which the config: and overrides: surfaces already are), not a list — so each element is addressed and replaced by key. Every resolved value maps 1:1 back to the single layer that supplied it, and channels resolve / explain --field report that layer.

Structural ops (overrides:) apply in a total order — layer precedence first, then declaration order within a layer. Collisions are errors, never silent no-ops: adding a node whose name already exists, or targeting a missing or already-removed node, fails with a diagnostic anchored to the offending op.

Overlays are resolved before executable compilation. Structural op streams are concatenated in total order and folded over the base node list. Clinker then compiles that target once for typed candidate validation; only when every config: candidate passes name, type, ambiguity, and fixed-lock checks does the winning config map enter executable compilation. Scoped vars: are likewise validated before executor initialization. One invocation produces one validated effective plan.

Value clobber: config, vars, and resources

The value-clobber surface carries scalar overrides. It appears identically on a group, a channel manifest, and a per-target overlay.

config: overrides composition config knobs, keyed by node.param dotted paths (the composition node’s name, then the parameter name):

config:
  scorer.threshold: { value: 0.95 } # override the `threshold` knob of `scorer`

The override changes executed behavior, not just the rendered provenance: the composition body reads the knob as $config.<param>, which the planner constant-folds to the resolved value for that instantiation at compile time. The winning layer is still recorded in the provenance side-table, so channels resolve / explain --field continue to report which layer supplied the value.

A config: key that matches no parameter in the compiled plan is a hard error (E113) — a misspelled or stale key aborts the run rather than silently doing nothing.

Rebinding a composition resource

resources: changes which logical catalog resource supplies one declared composition slot. The key is composition-node.slot; the leaf uses the same { value, fixed } shape as other clobbers:

# channel.cfg.yaml, a group file, or a per-target file
resources:
  lookup.orders: { value: tenant_orders }

The base composition call must already declare orders under _compose.resources_schema, and tenant_orders must exist under [catalog.resources] with the required kind and capabilities. Group, channel-wide, and per-target candidates use the ordinary precedence order and retain every attempted layer plus the winner. fixed: true locks a lower binding against higher layers just as it does for config values.

Only a scalar logical identity is accepted as value. Inline descriptors and credential/profile/secret/token selectors are rejected at the strict YAML leaf. An overlay cannot introduce a slot, address an internal nested slot, or change ports, composition names, or config through this surface. Resource rebinding changes the semantic plan fingerprint but does not resolve credentials or open runtime handles.

Locking a value: fixed

fixed is metadata on the value it locks, never a sibling map. A config leaf uses { value, fixed }; a variable leaf uses { type, default, fixed }. fixed defaults to false. Unknown spellings and a misplaced top-level fixed: block fail at the authored line with the corrected leaf form.

# channel.cfg.yaml — the channel-wide manifest
channel:
  name: tenant.globex
  targets: [daily.orders]
config:
  scorer.threshold: { value: 0.9, fixed: true }

# order_fulfillment.channel.yaml — the per-target overlay (a higher layer)
channel: { target: daily.orders }
config:
  scorer.threshold: { value: 0.95 } # rejected: channel-wide locked this key

The per-target candidate is invalid because the channel-wide value is fixed; the diagnostic points to the per-target leaf and the run does not start. The resolved provenance remains 0.9, and channels resolve marks that winning layer (fixed). Invalid candidates are validated even when another layer would win, so a typo or type mismatch cannot hide behind precedence.

vars: overrides or adds scoped-variable defaults, using the same four scopes a pipeline’s own vars: block uses ($vars.* / $pipeline.* / $source.* / $record.*). Each leaf is the same { type, default } shape a pipeline declaration uses:

vars:
  static:                    # $vars.*
    currency: { type: string, default: "USD", fixed: true }
  pipeline:                  # $pipeline.*
    cutoff_date: { type: date, default: "2026-01-01" }
  source:                    # $source.<src>.*  — outer key is the source-node name
    orders:
      ingest_label: { type: string, default: "prod" }
  record:                    # $record.*
    tier: { type: string, default: "bronze" }

See Variables for the scoped-variable model these overlay.

Structural ops: overrides

The overrides: surface is an ordered list of discrete, name-addressed ops applied to the base pipeline’s node list before compilation. Each op is a mapping with an op: discriminant. Unknown keys, or keys that belong to a different op kind, are rejected at parse time.

The op vocabulary is add / remove / replace / set / bypass / patch_schema.

add — splice in a node

Insert a new node, either inline or as a composition reference. The splice anchor is exactly one of after: / before: / an explicit input:.

overrides:
  # Inline transform, spliced after `normalize` (its former consumers now read `stamp`):
  - op: add
    node:
      type: transform
      name: stamp
      input: normalize
      config:
        cxl: "emit order_id = order_id"
    after: normalize

  # A composition, named by `alias`, with a config knob for the injected node:
  - op: add
    composition: ../composition/fraud_check.comp.yaml
    alias: fraud_check
    after: normalize
    config:
      threshold: 0.8

after: X reads from X and repoints X’s former consumers onto the new node; before: X feeds X, taking over X’s former upstream. An inline node with no splice anchor keeps its own declared input:. Adding a node whose name already exists is an error.

remove — delete a node and rewire

Delete a node by name, repointing its named consumers through an explicit rewire: map so no dangling reference is left behind:

overrides:
  - op: remove
    target: legacy_audit
    rewire:
      route_priority.input: product_lookup   # <consumer>.input: <new upstream>

Each rewire: key is a <node>.input path; each value is the replacement upstream. Any consumer still referencing the removed node afterward is an error, as is removing a node that does not exist.

bypass — remove a linear node

Sugar for remove on a 1-in/1-out node: it auto-rewires the node’s sole consumer onto its sole upstream.

overrides:
  - op: bypass
    target: legacy_audit

bypass only applies to a single-input, single-consumer node; a fan-in/fan-out node must use the explicit remove op with a spelled-out rewire: map.

replace — swap a node’s definition

Replace a whole node by name, keeping its identity (and therefore every consumer edge) intact. The replacement node’s own name: must equal target:.

overrides:
  - op: replace
    target: normalize
    node:
      type: transform
      name: normalize
      input: orders
      config:
        cxl: "emit order_id = upper(order_id)"

set — set one field within a node

Set a single field within a named node by path. The currently addressable path is config.cxl — the primary CXL body of a transform / aggregate / combine node — so replacing a stage’s logic wholesale is a set, not a special case:

overrides:
  - op: set
    target: route_priority
    field: config.cxl
    value: >
      emit _route = if priority_level == "urgent"
        then "priority_report" else "fulfilled_orders"

Here _route is an ordinary audit field; it does not select an Output. Direct Outputs sharing route_priority each receive every record. To partition rows by destination, add a Route node with conditions that read the field (or express the conditions directly on the Route).

Any other field path is a hard error, never a silent no-op.

patch_schema — shape a source’s columns

Add / rename / modify / remove columns on a source node’s declared schema, via a column-name-keyed map (the map key is the column name). Each column carries exactly one op:

overrides:
  - op: patch_schema
    target: orders
    schema:
      amount:      { type: float, scale: 2 }       # modify: set any subset of attrs
      cust_id:     { rename: customer_id }         # rename (a physical->logical alias)
      order_notes: remove                          # drop an existing column (bare scalar)
      region:      { add: { type: string } }       # add a new column (map key = new name)

The modify leaf is a bare attribute map: it sets any subset of the column’s attributes (type, scale, precision, format, width, …), leaf-replace, keeping every attribute it does not name. A typo’d attribute is rejected rather than silently appended. The same grammar applies identically at every override layer (pipeline / group / channel).

The keyed-map shape (rather than a list) is deliberate: a column op is addressed and leaf-replaced by name, with first-class rename / remove / add, exactly matching the source-config schema patch grammar so the two surfaces resolve columns and their diagnostics identically.

rename is a source-column alias, not a bare relabel: the reader still binds the original physical column and re-labels its value under the new name, so downstream CXL and the output see the new name carrying the original column’s data. A missing column, an add that collides with an existing name, or a rename onto an existing name are all errors (E231–E233).

To see which layer set a given attribute on a patched column, trace it with clinker explain <pipeline> --field <source>.<column>.<attribute> (optionally --channel <name>); the output names the winning Base < Pipeline < Group < Channel layer and each shadowed one. See Field provenance.

Groups and selectors

A group (group/<name>.group.yaml) is a reusable overlay layer that sits between the pipeline default and the channel layers. It carries the same two surfaces every layer carries — config: / vars: value clobber and an overrides: op list:

group:
  name: enterprise
  targets:
    pipelines: [daily.orders]
    compositions: [etl.normalize]
  match: 'tier == "enterprise"'   # optional selector; higher priority wins
  priority: 20
config:
  scorer.threshold: { value: 0.8 }
overrides:
  - op: add
    node:
      type: transform
      name: fraud_stamp
      input: normalize
      config:
        cxl: "emit order_id = order_id"
    after: normalize

A group plays two roles under one concept:

• Selector-derived — when match: is present, the group is applied automatically to every channel whose labels satisfy the CXL boolean. Multiple matching groups are ordered by priority (higher wins; the default priority is 0).
• Standalone / explicit — when match: is absent, the group is never auto-selected; it applies only when invoked by name with --group. Groups are channel-agnostic — their overrides never read channel labels — so any group can run standalone against the base pipeline, with or without a channel.

Every group owns a non-empty explicit targets: set of catalog pipeline and/or composition ids. A selector only narrows that set: a matching label can never make the group global. Forced --group use is target-bounded by the same set.

Selectors are label-only CXL

match: is a bare CXL boolean expression evaluated in a restricted label-only context: the only names in scope are the channel’s labels. $record / $source / $pipeline / $vars / $doc, window and aggregate calls, now, and wildcards are all rejected, so a selector is a pure, deterministic predicate over labels.

match: 'region == "west" and tier == "enterprise"'

Labels are typed from their YAML/JSON scalar kind (string, bool, int, float), so the typechecker rejects label/literal type mismatches. A selector that references a label a channel does not declare is a hard error, never a silent false — a typo surfaces as an unresolved-identifier error rather than quietly excluding the channel.

The channel manifest

channel.cfg.yaml declares the channel identity, its non-empty pipeline target set, identity labels, and optional channel-wide values:

channel:
  name: tenant.globex
  targets: [daily.orders]
labels: { region: west, tier: enterprise }   # identity — drives group selectors
config:
  scorer.threshold: { value: 0.9, fixed: true }
vars:
  static:
    currency: { type: string, default: "USD", fixed: false }

Labels are identity, never a pipeline override. The manifest and its target set are required. Channel-wide overrides: and sources: are forbidden because they would apply graph/source/schema changes without a single admitted target; move those operations into the corresponding target file.

The per-target overlay

A target file overlays exactly one manifest-declared catalog pipeline and its admitted composition closure. The channel.target: logical id is authoritative; the filename has no identity semantics:

channel:
  target: daily.orders
config:
  scorer.threshold: { value: 0.95 }
overrides:
  - op: patch_schema
    target: orders
    schema:
      tax_exempt: { add: { type: bool } }

Complete admission and execution identity

Channel loading is fail closed. Clinker canonicalizes the workspace root and candidate path, rejects traversal and symlink escapes, opens each admitted file once, verifies its post-open identity, and reads it into one bounded byte buffer. UTF-8 validation, parsing, and content identity all use that exact buffer; the bytes cannot be swapped between validation and hashing.

Planning validates the whole admitted catalog before selecting a requested pipeline or target:

• every manifest target must have exactly one target overlay;
• every declared target is parsed and validated, including targets not selected for this run;
• the complete reachable pipeline and composition closure is validated for every target; and
• group discovery, channel discovery, or file I/O errors abort admission rather than silently skipping an entry.

The planned execution identity includes the selected pipeline bytes and every applied layer in precedence order: defaults, ordered groups, the channel-wide overlay, and the selected target overlay. Group priority, declaration sequence, and whether membership was derived or explicit are part of that identity. Changing applied bytes or their order changes the identity; changing an unapplied overlay does not.

CLI surface

Running with overlays

# Run as a tenant: resolves catalog identities and derives target-bounded groups.
clinker run pipeline/order_fulfillment.yaml --channel globex --base-dir .

# Force-include a group by name, with or without a channel.
clinker run pipeline/order_fulfillment.yaml --group enterprise --base-dir .

run resolves the overlay stack from the workspace (rooted at --base-dir, default the current directory) and folds the resolved overrides into the plan before execution. Overlay flags shared across run and explain:

explain --field <node.param> --group <NAME> reports the same overlay stack for provenance lookups, mirroring run.

Inspecting overlays

channels resolve renders the effective post-overlay DAG for one target under a chosen channel and/or groups, with per-value provenance — which layer supplied each value and which group injected which node:

# Resolve the effective plan for the globex channel (derives matching groups from its labels)
clinker channels resolve pipeline/order_fulfillment.yaml --channel globex --base-dir .

# Preview a group overlay standalone (no channel)
clinker channels resolve pipeline/order_fulfillment.yaml --group enterprise --base-dir .

Here --channel is a logical id from [catalog.channels]. The selected pipeline must likewise appear in [catalog.pipelines]; resolve never guesses identity from the filename. Matching groups are considered only after their explicit target sets admit that pipeline or one of its composition dependencies.

channels lint compiles every cataloged channel target and reports every failure through the same resolver used by run and explain:

clinker channels lint --base-dir .

Membership and labels

# List the channels a group's selector currently matches
clinker channels group members enterprise --base-dir .

# Stamp/overwrite a label across one or more channels (idempotent)
clinker channels label set tier=enterprise globex initech --base-dir .

channels label set takes a key=value assignment; the value is typed by YAML scalar inference (true/false → bool, integers → int, decimals → float, else string) so numeric and boolean labels compare correctly against selectors. The channel manifest must already exist with its explicit channel.targets list; the command never creates a targetless manifest.

Renaming a base node

refactor rename-node renames a base node and propagates the rename to every overlay that references it (splice anchors, target:, rewire: keys) across the workspace:

# Preview every file that would change
clinker refactor rename-node pipeline/order_fulfillment.yaml orders purchases --dry-run

# Apply it, then re-lint
clinker refactor rename-node pipeline/order_fulfillment.yaml orders purchases --base-dir .
clinker channels lint --base-dir .

The new name must be letters, digits, and _ only.

Source config patches

Independent of the overlay op engine, a channel file can patch a source node’s parsed config directly through a sources: block, applied before validation and compile so the run behaves exactly as if the source YAML had been hand-edited. This is the same column-keyed schema grammar patch_schema reuses, plus multi-value and per-format option patches.

sources:
  transactions:                            # source-node name (unknown -> E230)
    options:
      record_path: batch_records           # set a scalar per-format option (bad key -> E235)
    split_to_rows:                          # keyed by field name
      items:      { mode: split, position_column: line_no }  # add-or-modify
      tags:       { position_column: ~ }    # clear one attribute
      line_items: remove                    # drop an entry (unknown field -> E234)
    max_output_rows_per_input: 10000        # replace the source ceiling; 0 disables it
    split_values:                           # keyed by field name
      codes:      { delimiter: "|" }        # add-or-modify an entry
      tags:       { delimiter: ~ }          # reset to the default delimiter
      notes:      remove                    # drop an entry (unknown field -> E234)
    schema:                                 # keyed by column name
      amount:      { type: float, scale: 2 }
      cust_id:     { rename: customer_id }
      order_notes: remove
      region:      { add: { type: string } }

All ops are keyed and leaf-replace — there is no deep-merge. On an existing split_to_rows / split_values entry a partial map is a modify: an omitted key keeps its current value, and a new entry takes the same defaults hand-written config would. Because an omitted key means “keep current”, clearing an attribute that is already set needs its own form — an explicit YAML null. On position_column that removes the attribute; on delimiter, which always holds some separator, it restores the ; default. options are merged onto the source’s current options and re-validated through the format’s option struct, so an unknown or mistyped key is rejected exactly as in hand-written config. A schema rename is a source-column alias — the same alias a base column can declare directly with source_name::

schema:
  # read the physical `cust_id` column, expose it downstream as `customer_id`
  - { name: customer_id, type: string, source_name: cust_id }

Format-structure patches (X12 / HL7 v2)

Beyond the format-agnostic ops above, a sources: patch can reshape the format-layer structures an X12 or HL7 source declares in its options: block — with keyed add/modify/remove grammar instead of blob-replacing the whole options map:

sources:
  interchange:                             # an X12 source
    group_section:                         # the GS functional-group declaration
      name: fg                             # rename the section (omit to keep)
      fields:
        e04: int                           # set/add a typed field
        e05: remove                        # drop a declared field
    set_section: remove                    # drop the whole ST declaration
  messages:                                # an HL7 v2 source
    split_fields:                          # keyed by positional field name
      f08: { components: 3 }               # add-or-modify a composite split
      f03: remove                          # drop a declared split

group_section / set_section patch the X12 nested-envelope declarations (the GS functional-group and ST transaction-set levels); split_fields patches the HL7 composite-field splits, keyed by positional field name and resolved by wire position (f8 and f08 address the same split). Each op applies only to a source of the matching format (anything else is E238). The set form is a partial modify on an existing declaration — an omitted name or axis width keeps its current value — and creates the declaration when absent, in which case name (X12) or components (HL7) is required (E240). Removing a declaration, field, or split the source does not carry is E239. These ops apply after the options merge, so they layer on top of an options value that replaces the same declaration in one patch.

Multi-record patches (discriminator-driven flat files)

A multi-record flat file interleaves several record layouts in one file, each identified by a discriminator tag. A sources: patch reshapes that layout with records: (keyed by record-type id) and a discriminator: merge, so a tenant’s record set can differ from the base without editing the pipeline:

sources:
  ledger:                                  # a multi-record source
    discriminator: { start: 2 }            # move the tag byte range (partial merge)
    records:
      detail:  { tag: X }                  # retag; a nested `columns:` reshapes fields
      trailer: remove                      # drop a record type
      header:                              # add a record type (map key = its id)
        add:
          tag: H
          columns:
            - { name: hdr_id, type: string, start: 1, width: 8 }

A records entry follows the same keyed grammar as schema: a bare remove drops the record type, { add: { tag, columns, ... } } declares a new one, and a bare attribute map modifies an existing one. A modify sets any subset of the record type’s tag / parent / join_key / description and carries a nested columns: map that runs the column-op grammar (modify / rename / add / remove) against that record type’s own fields. The discriminator: op merges field by field onto the current discriminator — a named field overwrites, an omitted one is kept — and the merged result must be a byte range (start + optional width) XOR a field.

These ops apply only to a multi-record schema (E241). Modifying or removing an unknown record-type id is E242, adding an id that already exists is E243, a merged discriminator that is neither pure byte-range nor pure field is E244, and a discriminator tag shared by two record types after the patch is E245.

Sources inside a composition body

A plain sources: key names a top-level source node. To patch a source declared inside a composition body, qualify the key with the composition call-site node name: <composition-node>.<source>. The composition body is expanded during compile, so the patch is applied to the body’s source when the body is bound — before the body typechecks — exactly as a top-level patch shapes a top-level source before it binds:

sources:
  enrich.lookups:                          # source `lookups` inside composition node `enrich`
    schema:
      code: { rename: lookup_code }

Resolution is one level deep: the qualifier must name a composition node in the pipeline (an unknown composition — or a nested a.b.c key naming a source inside a nested composition body — is E230), and the source half must name a source node declared in that composition’s body (an unknown one is E230, naming the body file). A plain unqualified key still targets a top-level source, and a name that matches no top-level source still fails with E230 — now hinting at the qualified form when the pipeline has compositions.

Note: an authored body Source must link to one declared composition resource slot with resource: <slot>; a direct path, glob, regex, or paths matcher is rejected. The source patch above changes schema/reader configuration but does not select the resource. Planning binds the slot and compiles a call-scoped Source instance. On a data run, a credential-free file binding opens through the CLI-prepared catalog factory and streams via the executor’s bounded Source path. Credential-bearing bindings remain unsupported and fail before opening because no profile-selection surface is available.

When a patch changes the effective source config, the run’s pipeline identity differs from the base and from other patched variants, so their outputs and lineage do not collide.

Diagnostics

Compositions

Compositions are reusable pipeline fragments that can be imported into multiple pipelines. They encapsulate common transform patterns – date derivations, address normalization, currency conversion – into self-contained, testable units.

Using a composition

A composition node in your pipeline references an external .comp.yaml file:

- type: composition
  name: risk
  input: orders
  use: "./compositions/risk_score.comp.yaml"
  inputs:
    inp: orders
  config:
    threshold: 0.5

The use: field points to the composition definition file. The inputs: map binds each declared composition input port to an upstream node. The top-level input: is also required by the current node shape, but inputs: is the authoritative port wiring. The config: block passes parameters that customize this invocation.

Resolving the use: path

A use: value names a .comp.yaml in the workspace. It is resolved relative to the directory of the pipeline file being compiled, then against the set of .comp.yaml files discovered under the workspace root, finally falling back to a filename match. A use: that resolves to no .comp.yaml — a typo, a wrong relative prefix, or a file that does not exist — fails compilation with a spanned E103 diagnostic naming the composition node. The whole run aborts loudly; it does not silently drop the composition and write an empty output. The same holds for the other composition-binding errors (E102–E108): an ill-bound call site fails compile rather than producing a run that writes zero records. Run clinker explain --code E103 for details.

Composition definition file

A .comp.yaml file declares its interface in _compose: and its executable subgraph in nodes::

# compositions/risk_score.comp.yaml
_compose:
  name: risk_score
  inputs:
    inp:
      schema:
        - { name: order_id, type: string }
        - { name: amount, type: float }
  outputs:
    out: scored
  config_schema:
    threshold:
      type: float
      default: 0.5
      range: [0.0, 1.0]

nodes:
  - type: transform
    name: scored
    input: inp
    config:
      cxl: |
        emit order_id = order_id
        emit amount = amount
        emit high_value = amount >= $config.threshold * 2000.0

Composition fields

Reading config parameters in the body

A composition body reads its own config parameters as $config.<param>. The planner constant-folds each reference to the value resolved for that instantiation — the call site’s config: value, or a channel/group config: override, or the declared default — so the same composition used with different config: compiles to different bodies. Because the resolution happens per instantiation, a channel or group config: override changes what the body computes, not just the reported provenance.

Explaining config provenance

Every resolved composition parameter retains its base value and each attempted group, channel-wide, and per-target override. The winning layer, shadowed values, fixed locks, and source spans remain attached to the stable compiled node identity. Inspect a value with either a unique shorthand or its exact versioned address:

clinker explain pipeline.yaml --field 'risk.threshold'
clinker explain pipeline.yaml \
  --field '/v1/config/nodes/risk/fields/threshold'

The output includes the canonical exact address and lists layers in a stable order. Exact addresses include every enclosing composition call. For example, two sibling calls may each contain a local node named shared:

/v1/config/calls/left/nodes/shared/fields/threshold
/v1/config/calls/right/nodes/shared/fields/threshold

In that case shared.threshold fails with E118 instead of selecting one by insertion order. The diagnostic lists both exact addresses in deterministic order; copy the intended --field correction. An unknown query fails with E117 and lists only same-field candidates, never unrelated nodes or fields. An empty query fails with E116.

Address segments use RFC 6901 escaping: ~ becomes ~0 and / becomes ~1. Unicode remains unchanged. This makes rendering and parsing lossless, including after provenance serialization and repeated inspection of a compiled plan.

Source-schema provenance uses the parallel exact form /v1/schema/sources/<source>/columns/<column>/attributes/<attribute>; the three-part source.column.attribute shorthand remains available.

Body validation

Nodes inside a composition body are validated with the same node-scoped config checks as top-level pipeline nodes. A body node that would be rejected at the top level — an envelope wiring the not-yet-supported trailer: port, a transform declaring a reserved variable name or a default that does not match its declared type, an invalid log directive, or a batch_size: 0 — fails compilation with an E115 diagnostic naming the composition call site, the body file, and the violation. Run clinker explain --code E115 for details.

A body source or sink that sets a CSV delimiter or quote_char which is not exactly one ASCII byte is likewise rejected at compile time, not first at run, with the same one-byte rule top-level nodes get.

A body source or sink whose schema: names an external .schema.yaml file has that path resolved relative to the composition file’s own directory (not the invoking pipeline’s), and the file’s columns are inlined before the body binds. A body Sink therefore rounds decimal columns to their declared scale at the write boundary exactly as a top-level Sink does.

A body sink cannot be combined with a Source that declares dlq_granularity: document: document-level dead-lettering needs every Sink at pipeline level, and the pipeline fails compilation with E378. Run clinker explain --code E378 for how to move the Sink out of the body.

Executable example corpus

The five fragments under examples/pipelines/compositions/ are executable examples of the current authoring surface:

• Clean Names
• Fiscal Date Fields
• Order Classification
• Shipping Cost
• Validate Email

Each fragment uses _compose.inputs, _compose.outputs, config_schema, and the unified nodes list. The two date-dependent examples require an explicit as_of_date configuration value so their results do not depend on the day the test runs.

The composition example test recursively inventories every .comp.yaml file in that directory. Its case manifest must name exactly that discovered set: empty or missing inventories, duplicate keys, missing or extra cases, and paths that escape the corpus directory fail with distinct diagnostics before any example runs. Each case is then loaded through the production composition loader, placed in a generated pipeline, and executed by the real clinker binary. The test checks the exit status, record counters, and output bytes.

clinker run --explain proves that a generated pipeline compiles. It does not prove that the example produces the documented result; the executable corpus’s byte comparison is the behavior check.

Advanced wiring

For compositions with multiple input ports, bind every declared port by name:

- type: composition
  name: enrich_address
  input: orders
  use: "./compositions/order_product_enrich.comp.yaml"
  inputs:
    orders: orders
    products: product_catalog

The primary input: must be present for the current YAML node shape. The planner builds composition edges from the named inputs: map, so that map must contain every required port declared by _compose.inputs.

Downstream nodes consume a composition’s declared output ports using the usual node.port syntax. If there is only one output port, the bare composition node name selects it.

Resource bindings

A composition declares the external capabilities it needs as typed slots. The currently admitted kind is file; it requires read capability and a run-local file opener:

_compose:
  name: order_lookup
  inputs:
    input: { schema: [{ name: order_id, type: string }] }
  outputs: { out: order_reference }
  config_schema: {}
  resources_schema:
    orders:
      kind: file
      required: true

nodes:
  - type: source
    name: order_reference
    config:
      name: order_reference
      type: csv
      resource: orders
      schema: [{ name: order_id, type: string }]

resource: is an explicit body-Source-to-slot link. The slot must be declared by the enclosing _compose.resources_schema; the Source name and its format do not select a resource implicitly. A resource-backed body Source must not also declare path, glob, regex, or paths, because the bound catalog resource is its only external target. Every authored body Source must declare resource:. Composition input ports are separate synthetic roots: to consume caller-provided rows, declare _compose.inputs.<port> and set a downstream node’s input: <port> instead of authoring a Source node for that port.

Top-level Sources are unchanged: they continue to require exactly one direct matcher. resource: on a top-level Source is rejected until a separate top-level binding surface is designed.

The workspace supplies a concrete, secret-free descriptor under a logical identity in clinker.toml:

[catalog.resources.shared_orders]
kind = "file"
path = "data/orders.csv"
access = "read"

The call site binds only the declared slot to that logical identity:

- type: composition
  name: lookup
  input: orders
  use: ./compositions/order_lookup.comp.yaml
  inputs: { input: orders }
  resources: { orders: shared_orders }

Resource descriptors are strict. Unknown kinds or fields, inline objects, unknown catalog identities, undeclared slots, missing required slots, and kind/capability mismatches fail planning. A call site cannot contain a path, credential profile, secret, token, or opened handle. File descriptors must remain inside the workspace and the catalog is admitted under fixed entry and descriptor-byte limits.

Planning retains the winning logical identity and every attempted overlay layer for each binding. That identity also participates in the semantic plan fingerprint. For each authored body Source, planning compiles a distinct call-site-scoped instance carrying only the slot, logical identity, resource kind, required capabilities, opener family, run lifetime, provenance, and stable logical dataset identity. It does not retain the catalog’s physical path. During clinker run, the CLI resolves that credential-free file requirement at the workspace edge and transfers an opaque single-use reader factory to the executor. The executor acquires the complete compiled group, opens all of its Sources before starting any of them, and streams their finite records through the ordinary bounded Source path. An open/read failure or interruption closes every opened session and releases the group. This surface still does not select credentials: a group requiring credentials fails before runtime effects because no credential-profile option exists yet.

Ordinary composition calls do not have outputs: or alias: fields. Either key fails with E377 at its authored location. Use _compose.outputs for the public output contract and the composition node’s name for its caller-visible namespace. add.alias remains valid only inside an overlay add operation, where it names the inserted node.

Call-site fields

Contract status

The bounded catalog, typed file slot/binding, overlay provenance, stable logical dataset identity, and E377 call-surface rejection implement the planning half of D-12 through D-16. Credential references, credential resolution, and runtime handle activation are not implemented; a resource binding is therefore a validated planning contract, not permission to perform I/O.

See the canonical composition-resource and call-site contract for status, evidence, compatibility impact, and the AUTH-01 boundary.

Complete example

pipeline:
  name: order_pipeline

nodes:
  - type: source
    name: orders
    config:
      name: orders
      type: csv
      path: "./data/orders.csv"
      schema:
        - { name: order_id, type: string }
        - { name: amount, type: float }

  - type: composition
    name: risk
    input: orders
    use: "./compositions/risk_score.comp.yaml"
    inputs:
      inp: orders
    config:
      threshold: 0.5

  - type: sink
    name: result
    input: risk
    config:
      name: result
      type: csv
      path: "./output/scored_orders.csv"

Correlation Keys

A correlation key declares a set of records from a single source as an atomic group: if any record in the group fails validation or processing, the whole group is sent to the DLQ. This is the right shape for transactional data where partial processing is worse than total rejection – the canonical example is an order with multiple line items where one bad line should reject the entire order.

This page describes how to declare a correlation key and how it behaves through each node that can fan out, fan in, group, or join records.

Interactive companion: the correlation keys explainer lets you make lines fail and see what reaches the output and the DLQ, with and without an Aggregate.

Declaration

Correlation keys are declared per source. Each source’s config: block carries an optional correlation_key: field naming the column (or list of columns) whose value identifies a record’s correlation group within that source.

nodes:
  - type: source
    name: orders
    config:
      name: orders
      type: csv
      path: ./data/orders.csv
      correlation_key: order_id
      schema:
        - { name: order_id, type: string }
        - { name: amount, type: int }

  - type: source
    name: customers
    config:
      name: customers
      type: csv
      path: ./data/customers.csv
      correlation_key: [customer_id, region]   # multi-column key
      schema:
        - { name: customer_id, type: string }
        - { name: region, type: string }
        - { name: name, type: string }

  - type: source
    name: sensor_readings
    config:
      name: sensor_readings
      type: csv
      # No correlation_key: record-level errors land in the DLQ as
      # standalone entries with no group atomicity.
      schema:
        - { name: ts, type: date_time }
        - { name: value, type: float }

A record’s correlation group is identified by the tuple of values for that source’s listed fields. Records sharing the same tuple within the same source belong to the same group. There is no pipeline-level correlation key — declare it on each contributing source.

The group identity is captured at ingest, so rewriting the key column in a later Transform does not change a record’s group — anonymizing or transforming order_id downstream still keeps the original grouping intact.

A source whose declared correlation_key: field names a column not present in its own schema: block is rejected at compile time with diagnostic E153. The fix is to add the field to the schema or remove it from correlation_key:.

Row order

Declaring correlation_key on a Source changes the order its rows reach the rest of the pipeline. When the Source’s declared sort_order already begins with the correlation-key fields, in either direction, that declared order is kept as it is. Otherwise the planner inserts a sort after the Source, ascending on the correlation key and then on the remaining declared sort_order fields, so that each group’s rows are adjacent. Rows with equal keys keep their order in the file.

Every consumer that depends on row order sees this order: Sink output order, an Aggregate’s first-arriving value, Cull and Reshape ties, and which build record a Combine’s match: first picks. Adding a correlation key to an existing pipeline can therefore change those results even when no record fails.

DLQ semantics

When a record fails inside a correlation group:

• The failing record produces a trigger DLQ entry. Its category reflects the actual failure (e.g. type_error, validation_failed).
• Every other record from the same source in that group produces a collateral DLQ entry, carrying the category correlated.
• Records belonging to other (clean) groups proceed normally.

A record with a null value for the correlation-key field is treated as its own group: it has no peers, so DLQ atomicity does not span multiple records.

The dlq_count counter sums triggers and collaterals. It counts one trigger per failure, so a row that fails twice in a group counts twice, as it does without a key.

Group buffering

The engine buffers records per correlation group until either the group completes or a failure triggers a flush. The max_group_buffer: field on the pipeline-level error_handling: block caps per-group buffering across every source’s groups:

error_handling:
  max_group_buffer: 100000     # Default: 100,000

A group that goes over the cap is dead-lettered whole when the run commits it. It is not a hard error: the run continues.

• Each row of the group that failed on its own is written as its own trigger, with its own category and _cxl_dlq_id.
• The group’s other rows are written under one group_size_exceeded trigger, the first of them; the rest are correlated rows that carry its id as their _cxl_dlq_trigger_id.
• A group whose rows all failed writes only those failures, with no group_size_exceeded row.

Each failure is written once, with its Combine build row if it has one (see Combine interaction), each condemned row once, and every row counts toward dlq_count and the DLQ rate limits. The group_size_exceeded row’s _cxl_dlq_timestamp is when the group went over the cap, and its error detail states the cap and how many entries the group held.

The cap counts the entries a group holds, not its distinct rows: a row counts once for each Sink it reaches, and each failure counts once. A row that an inclusive Route sends to two Sinks counts twice, and so does a row that fails on one branch and reaches a Sink on another. A failing Combine match counts once, although it holds both the driver row and the matched build row.

Going over the cap does not stop a group from buffering. The group keeps buffering its rows until the run commits it, so today the cap decides how a large group is dead-lettered but does not bound the memory it uses.

Per-operator interactions

Route interaction (fan-out)

A correlation group can span multiple route branches. Group atomicity is preserved across branches: if any record in the group fails (in any branch’s transform, or in the route predicate itself), the entire group is rejected from every branch.

For an inclusive route where one record reaches both branches, a single failure DLQ’s that source row exactly once — not once per branch. A record that fails on both branches has two failures and is written twice, once as each branch’s trigger with its own stage, as it is without a key.

Merge interaction (fan-in)

Merge concatenates upstream branches that share a schema. Records keep their correlation identity through the merge, so rows from different sources that share the same key value become one correlation group downstream: a failure on any one of them DLQ’s the whole group across both sources.

Per-source rollback narrowing

When two sources contribute records to the same correlation group, a failure originating from one source does not collaterally DLQ records from the other source. The collateral fan-out is scoped to the failing source’s records only.

For example, with [src_a, src_b] → merge → transform → out where both declare correlation_key: id, an error that fires on a src_b row produces a trigger for that row while the src_a row sharing the same id is spared and reaches the output. Single-source pipelines behave exactly as a pipeline-wide collateral DLQ would, since every co-grouped record shares the one source.

Two cases stay group-wide rather than narrowing per source:

• max_group_buffer overflow DLQ’s every record in the overflowing group — no single source is to blame for the overflow.
• Combine output failures DLQ the synthesized output row, which has no single-source attribution. The exception concerns the output row only: the matched build record’s dead letter follows the failing driver’s group, as described under Combine interaction, and never widens the narrowing to the build record’s source.

Aggregate interaction

When an aggregate’s group_by covers every correlation-key field, the aggregate stays on the strict path: each emitted row inherits the correlation identity of its inputs, and any DLQ trigger in the group rolls back every record in the group, including the aggregate output row.

- type: aggregate
  name: order_totals
  input: orders                         # correlation_key: order_id
  config:
    group_by: [order_id]                # covers the key
    cxl: |
      emit total = sum(amount)

When an aggregate’s group_by omits a correlation-key field, the engine automatically retracts only the failing records and recomputes the affected groups, so the surviving contributions still produce a correct aggregate row. You do not configure this — the engine picks the path from the group_by content. (One restriction: this mode cannot be combined with strategy: streaming, which is rejected at compile time.)

- type: aggregate
  name: dept_totals
  input: orders                         # correlation_key: order_id
  config:
    group_by: [department]              # omits the key — surviving rows recomputed
    cxl: |
      emit total = sum(amount)

Combine interaction

Every combine declares propagate_ck: to select which correlation-key fields its output rows carry:

• propagate_ck: driver — output inherits only the driver input’s correlation identity. The common case; today’s strict-correlation pipelines stay on this setting.
• propagate_ck: all — output carries the union of correlation-key fields across every input. Use when the build side carries keys that downstream operators need to read.
• propagate_ck: { named: [<field>, ...] } — output carries exactly the named subset. Use to project a multi-field key down after a join.

- type: combine
  name: enriched
  input:
    o: orders                          # driver (correlation_key: employee_id)
    d: departments                     # build side
  config:
    where: "o.employee_id == d.employee_id"
    match: first
    on_miss: skip
    cxl: |
      emit employee_id = o.employee_id
      emit amount = o.amount
      emit dept = d.dept
    propagate_ck: driver

How match mode fills the propagated key:

• match: first — the single matched build’s key fills the slot.
• match: all — one output row per matched build, each carrying its own build’s key.
• match: collect — one row per driver; the first matched build’s key fills the (single-valued) slot, while every matched build’s full payload still rides inside the array column.

Driver wins on a name collision: if both the driver and a build input declare the same key field, the output keeps the driver’s value.

propagate_ck is a required field — every combine must spell out which mode it uses.

A failing match’s build-side dead letter follows the driver’s group. When the combine body fails for a driver row, that driver row is the trigger of the driver’s correlation group. The matched build record’s dead letter is held with the same group as a collateral (_cxl_dlq_trigger: false, category combine_output_row), written right after its driver’s row and carrying that driver’s _cxl_dlq_trigger_id, and it is written or rolled back exactly when the driver’s group is. It never condemns the build record’s own correlation group: another driver that matched the same build record keeps its output unless its own group failed. The build record is written once per failure: when several drivers fail against one build record, in one group or in several, each failing driver’s row is followed by its own copy of the build row, carrying that driver’s _cxl_dlq_trigger_id. A driver that fails against several build rows is written once per failure, each copy followed by the build row of that failure.

Composition interaction

A composition’s body operates on records flowing in from the parent pipeline; correlation identity flows into the composition inputs and back out the named ports unchanged. Compositions cannot declare their own correlation key — a key is a property of a source, not of the composition body that consumes a source’s records.

Debugging

Correlation grouping is tracked on internal columns you never write in YAML or CXL, and they are hidden from writer output by default. To surface them for debugging, set include_correlation_keys: true on a Sink node:

- type: sink
  name: debug
  input: any_node
  config:
    type: csv
    path: "./debug.csv"
    include_correlation_keys: true

The output then contains extra columns named $ck.<field> (literal prefix in the CSV header) for each declared correlation-key field.

To investigate DLQ collaterals: every collateral entry’s category is correlated, and the trigger entry in the same group carries the actual failure category and message.

See also

• Error Handling & DLQ – general DLQ configuration, fail-fast vs continue, type-error thresholds.
• Aggregate Nodes – group-by semantics and the strategy hint.
• Combine Nodes – driver selection and match modes.
• Sink Nodes – include_correlation_keys and other field-control flags.

Document Envelope Context ($doc.*)

Many enterprise file formats wrap their record body in an envelope: named sections that surround the records and carry document-level metadata — a batch header with a run date and batch id, a trailer with a record count and checksum, or arbitrary sibling sections. Clinker exposes these sections to CXL through the $doc.<section>.<field> namespace.

nodes:
  - type: source
    name: payments
    config:
      name: payments
      type: xml
      path: data/payments.xml
      options:
        record_path: payments/Payment
      envelope:
        sections:
          BatchInfo:
            extract: { xml_path: "/payments/BatchInfo" }
            fields:
              batch_id: string
              run_date: date
          Summary:
            extract: { xml_path: "/payments/Summary" }
            fields:
              record_count: int
              checksum: string
      schema:
        - { name: amount, type: int }

Interactive companion: the document context explainer shows which records read each section, how each file becomes its own document, and what dlq_granularity: document rejects.

Like the rest of the pipeline config, the envelope: block is strict: an unknown key at any level (a misspelled sections:, extract:, or fields:) is rejected at plan parse time with a diagnostic naming the bad key, rather than being silently ignored.

A downstream transform reads any declared section field on every body record:

nodes:
  - type: transform
    name: tag
    input: payments
    config:
      cxl: |
        emit batch = $doc.BatchInfo.batch_id
        emit expected_total = $doc.Summary.record_count
        emit amount = amount

Section names are yours

The engine reserves no section names. BatchInfo and Summary above are arbitrary identifiers chosen by the pipeline author — Head / Foot, preamble / trailer, batch_metadata / eob_summary are all equally valid. A section name is whatever string you put in the sections: map; CXL exposes it verbatim as $doc.<that_name>.<field>.

Extracted sections are available throughout the body stream

Every extracted section is available to every body record. JSON and XML can pre-scan declared sections anywhere in the document, so a header at the top and a trailer at the bottom are both visible from the first record to the last. SWIFT likewise scans its service blocks before emitting fields. Multi-record CSV and fixed-width record_type extraction captures the leading header region only; it does not extract arbitrary trailing sections.

For formats with trailing-section extraction, a trailer field is available during body processing, not just at end-of-file. A Transform can compare every row, including the first, against the trailer’s count:

- type: transform
  name: check_count
  input: payments
  config:
    cxl: |
      emit amount = amount
      emit declared_count = $doc.Summary.record_count

Note that an extracted trailer section you read via $doc.* is distinct from the structural counts an EDI reader validates internally (the X12 SE/GE/IEA, EDIFACT UNT/UNZ, HL7 BTS/FTS segment counts). Those trailer counts are checked by the reader against the body it streamed, and a mismatch is a structural-integrity failure — see Malformed envelopes for how dlq_granularity: document dead-letters a malformed file instead of aborting the run.

The pre-scan reads the envelope-bearing segments of the file before body streaming begins. The amount of retained data depends on the declared sections, their payloads, and the format’s path to a trailing section:

• JSON streams the pre-scan. The reader walks the document once and deserializes only the subtrees the declared sections point at — every other key (including a multi-megabyte body array) is parsed-and-skipped without being stored. The retained sections live in a bounded document index capped by max_index_bytes (see below), so retained section memory scales with the declared sections, not the document size.
• XML streams the pre-scan too. The reader event-walks the document once and flattens only the declared section subtrees — every other element, including a multi-megabyte body, is event-walked and dropped without being materialized. The retained sections live in the same bounded document index capped by max_index_bytes. The reader still holds the file’s raw bytes for the lifetime of the read (one disk read backs both the body parser and the pre-scan), but the pre-scan no longer materializes the undeclared section subtrees.

For multi-record CSV, retained section names, field names, values and document containers carry allocation ownership against the run’s finite memory budget. Their charge lasts through downstream document use and any surviving aliases. A repeated header does not create an unbounded history of section snapshots; resource refusal remains fatal even under strategy: continue. CSV accepts the same UTF-8 and Latin-1 policy for section cells as for body cells.

Only the envelope sections live in the document context. This separation does not mean the executor retains only one body record: downstream Source and Combine paths can still materialize whole inputs. See what the memory budget measures. Other formats’ existing document-index limits remain as described below.

Bounding envelope retention with max_index_bytes

For JSON and XML sources, the document index is capped so a pathologically large declared section fails loud rather than exhausting memory. The cap is charged incrementally as each section is parsed — byte by byte while the section’s subtree is built — so even a single oversized declared section aborts mid-parse, before its whole subtree materializes, naming the offending section and the cap. (Undeclared siblings, including a multi-megabyte body, are skipped without being parsed into the index at all.)

- type: source
  name: events
  config:
    name: events
    type: json
    path: "./data/events.json"
    options:
      record_path: data.rows
      max_index_bytes: 64MB   # cap on retained envelope sections

The same max_index_bytes option applies to an XML source’s options: block, capping its envelope pre-scan identically.

max_index_bytes accepts a decimal size string (64MB, 500KB) or a bare byte count. It is optional; when omitted the reader applies a documented finite default of 64MB. Only the declared sections a program actually reads are retained, so envelope metadata sits far below this ceiling in practice — the cap exists to convert an unbounded mistake into a clear error.

Extract rules per format

Each section declares how the reader locates its payload:

xml_path and the source-level record_path option are both slash-paths over XML but root differently: xml_path tolerates a leading / (/doc/Head is its documented form), while record_path rejects one. They locate different things and are deliberately not aligned — see XML Format → record_path and xml_path root differently.

Declaring an xml_path section against a JSON source (or vice versa), a segment extract against XML/JSON, a record_type extract against any format other than multi-record CSV / fixed-width, or any envelope: block at all on a plain (single-schema) CSV or fixed-width source is a configuration error that fails fast rather than silently producing empty sections.

A json_pointer must be a valid RFC 6901 pointer: either empty ("", naming the whole document) or a /-introduced path such as /Head or /batch/summary. A slashless value like Head is a typo — it would decode to zero segments and silently match the root document — so it is rejected at validation rather than resolving to the wrong metadata.

A plain (single-schema) CSV or fixed-width source carries no envelope — there is no header/trailer structure to extract. Declaring envelope: sections on one is a configuration error (E356): the sections would never be populated and every $doc.<section>.<field> against the source would resolve to null, so the compiler rejects it rather than accepting an inert declaration. Envelope extraction on a flat file — the record_type extract — applies only to a multi-record source, one declaring a discriminator: + records: block; declare that schema if the file genuinely carries header/trailer records.

Network (REST) sources carry no $doc context

A rest source pulls its records page by page over paginated HTTP — it has no single buffered document with head and tail sections, so it carries no envelope context. Envelope sections are a file-document concept, so the compiler rejects them on a REST source rather than letting them silently resolve to null:

• Declaring an envelope: block on a rest source is an error (E349) — the declaration would be inert.
• Reading $doc.<section>.<field> from a node fed by a rest source is an error (E349) — the access can never resolve.

Pull the document-level metadata into record fields through the API’s own response shape (record_path, split_to_rows) instead, so the value travels as a normal field rather than as document-envelope context.

EDIFACT segment extract

An EDIFACT source exposes its interchange header UNB as an envelope section. The section’s field names are the positional element keys e01, e02, … :

envelope:
  sections:
    interchange:
      extract: { segment: "UNB" }
      fields:
        e05: string          # interchange control reference

Only the UNB header is extractable. Trailer segments (UNT, UNZ) that arrive after the body are not envelope sections — their control counts are validated by the reader instead. A mismatch between a trailer’s declared count and the body the reader streamed is a structural-integrity failure: by default it aborts the run, and under a source’s dlq_granularity: document opt-in it dead-letters the whole file to the DLQ (see Malformed envelopes). A segment extract naming any tag other than UNB is rejected at startup. See EDIFACT Format for the full reference.

Multi-record record_type extract

A multi-record CSV or fixed-width source — one that declares a discriminator: and a records: list — exposes a header record type as an envelope section through the record_type extract. The tag names which of the source’s declared record types carries the section’s payload; the matched header row’s named fields become the section’s fields.

schema:
  discriminator: { start: 0, width: 1 }
  records:
    - { id: header,  tag: H, columns: [ { name: batch_id, type: string, start: 1, width: 9 } ] }
    - { id: detail,  tag: D, columns: [ { name: amount,   type: int, start: 1, width: 9 } ] }
    - { id: trailer, tag: T, columns: [ { name: count,    type: int, start: 1, width: 9 } ] }
  structure:
    - { record: trailer, count: count }
envelope:
  sections:
    head:
      extract: { record_type: H }       # the H header row → $doc.head.*
      fields:
        batch_id: string

Only a header record type — one whose rows precede the body at the file head — is extractable as a $doc section; the reader captures the first such row in a bounded pre-scan and excludes it from the body stream. A trailer record type (one named by a structure: constraint) arrives after the body it closes, so it is not an envelope section — its declared count is validated against the streamed body count instead, the same structural-integrity check the EDI trailers use. See CSV and Fixed-Width for the full reference.

A JSON example:

- type: source
  name: payments
  config:
    name: payments
    type: json
    path: data/payments.json
    options:
      record_path: records
    envelope:
      sections:
        Head:
          extract: { json_pointer: "/Head" }
          fields:
            batch_id: string
        Foot:
          extract: { json_pointer: "/Foot" }
          fields:
            count: int
    schema:
      - { name: amount, type: int }

against:

{
  "Head": { "batch_id": "RUN-001" },
  "records": [ { "amount": 10 }, { "amount": 20 } ],
  "Foot": { "count": 2 }
}

Typed fields

Each section’s fields: map declares the field name and its type, drawn from the same small vocabulary as source schemas: string, int, float, bool, date, date_time. The extracted raw value is coerced to the declared type at pre-scan time; a value that cannot coerce (e.g. a non-numeric string declared int) fails the source with a diagnostic naming the section, field, and offending value.

A field that the document does not carry resolves to null — $doc.* follows the same missing-value convention as $source.* and $pipeline.*. A section that the document does not carry at all is simply absent from the context; any $doc.<missing_section>.<field> resolves to null.

Declared-path validation

Every $doc.<section>.<field> reference is cross-checked at compile time against the schema the feeding source’s reader will actually serve. A reference that can never resolve — almost always a typo — is rejected at compile time, pointing at the node that made it, rather than resolving silently to null. How the path is checked depends on the source:

Closed-schema sources (XML, JSON) — the envelope: block is the complete schema: the reader extracts exactly the sections and fields it declares. A reference naming a section the source does not declare, or a field the declared section does not declare, is rejected with error E341 ($doc.Summry.total against a declared Summary). Run clinker explain --code E341 for the full write-up.

Multi-record CSV / fixed-width sources — a source whose schema declares discriminator: + records: exposes a header record type as a $doc section through the record_type extract. The reader coerces the matched header record’s columns through the section’s declared fields: and serves exactly those fields, so the section is closed just like an XML/JSON one: a reference naming an undeclared section, or a field the section does not declare, is rejected with error E341. (A plain single-schema CSV / fixed-width source has no such structure — declaring an envelope: on one is rejected with error E356.)

Segment/positional sources (X12, EDIFACT, HL7) — the file-level header (ISA/UNB/FHS) is declared through envelope: and is closed, but the reader also synthesizes nested envelope levels the config never names — X12’s functional_group / transaction_set, HL7’s batch / transaction_set — keyed by positional eNN / fNN elements bounded by the source’s max_elements / max_fields. A $doc path is checked against that synthesized vocabulary plus any section/field you declared, so a legitimate wire-derived path ($doc.functional_group.e06) is accepted while a misspelled section ($doc.functonal_group.e06) or an out-of-range positional element ($doc.transaction_set.e99) is rejected with error E348. Run clinker explain --code E348 for the full write-up.

REST sources carry no document and reject $doc outright — see Network (REST) sources carry no $doc context above (E349).

Plain (single-schema) CSV / fixed-width and SWIFT MT sources are not statically checked. A plain flat file synthesizes no $doc sections, and SWIFT serves declared sections under user-chosen or default block labels — neither fits the closed or positional model cleanly. (A multi-record CSV / fixed-width source is checked — see above.)

Indexed $doc access

A section field that holds an array or a map can be indexed inline, the same way any record value is — see Nested paths for the full bracket-index reference. Integer indices select array elements; string keys select map entries; the two compose into a chain:

cxl: |
  emit first_line = $doc.Header.line_items[0]          # array element
  emit run_date   = $doc.Header.meta["run_date"]       # map entry
  emit first_sku  = $doc.Header.line_items[0]["sku"]   # array-of-maps chain

An out-of-range array index or a missing map key resolves to null — it never errors or panics, and a null mid-chain short-circuits the rest of the chain to null rather than failing. This is the same missing-value convention $doc.<missing_field> and $source.* follow.

Only literal paths are readable

Every index segment must be a literal — a constant integer or string written in the program text. The section and field are always literal identifiers (the grammar requires it), so a literal index is the last piece a reader needs to know, before reading any input, exactly which envelope paths a run will consume. The pre-scan extracts precisely those statically-resolvable paths and nothing else.

A computed index — one derived from runtime data, such as $doc.Header.line_items[row_index] — is not statically resolvable: the reader cannot pre-scan a row-dependent element. clinker rejects it at compile time with a diagnostic pointing at the offending index, rather than reading it at run time. Use a literal index, or pull the value into a record field upstream and index that instead.

This compiles together with the declared-path rule above: a $doc read of a section or field the source does not declare is a compile-time error (E341), and a $doc read with a computed index is a compile-time diagnostic — neither reaches run time as a silent null. Only a literal path over a declared section is pre-scanned and readable.

One document per file

Each source file is its own document with its own envelope context. When a source matches multiple files (via glob: / paths:), each file gets a fresh document context with its own section values. Records from different files never share a context — a record’s $doc.* always reflects the file that record came from.

Document boundaries flow through the pipeline so that document-scoped operators fire at exactly the right point. A document-scoped operator fires exactly once per document, even when that document arrives across several inputs that a Merge or Combine brings together.

Per-document aggregation

A grouped or global Aggregate reading a multi-document source produces one set of grouped rows per document, not a single aggregate spanning every document. When a document closes, the Aggregate finalizes and emits the groups belonging to that document, then drops their state before the next document accumulates — so a glob: source over twelve monthly files through a group_by Aggregate yields twelve independent monthly roll-ups, and only one document’s groups are ever live at once (the others have already been emitted and freed, or have not yet started).

This applies only when a document boundary actually reaches the Aggregate. A plain single-file source is one document, so it still emits one aggregate. A Merge that combines several distinct single-document sources flushes those sources independently downstream — one roll-up per source document, exactly as feeding each source to its own Aggregate would. This holds for every Merge mode.

A Combine (join) preserves document boundaries on every strategy, so a per-document Aggregate downstream of a join also rolls up per driver document.

Nested (multi-level) envelopes

Some formats wrap their records in several envelope levels, one inside another. EDI X12 is the canonical example and the first format that implements this: an interchange (ISA/IEA) contains one or more functional groups (GS/GE), each containing one or more transaction sets (ST/SE), each containing the records. A single file can carry multiple interchanges back to back. See X12 Format for the full reference.

HL7 v2 is the second multi-level format: an optional file (FHS/FTS) contains optional batches (BHS/BTS), each containing one or more messages (MSH..), each containing the segment records. The tiers map onto the same nested levels — the FHS file header is a declared segment: "FHS" section, while the BHS batch and the MSH message surface automatically as the reader-supplied sections batch and transaction_set. Every tier is optional, and a level’s section exists only when its header segment is present in the input — the reader synthesizes a batch section only where it reads a BHS, and the message level only where it reads an MSH.

A bare MSH-led stream — messages with no BHS batch and no FHS file wrapper — therefore opens only the message level. $doc.transaction_set.* resolves against each message, but $doc.batch.* resolves to null: no BHS was read, so no batch section was synthesized. Whether a file carries a BHS batch wrapper is a property of the input bytes, not the config, so a $doc.batch.* path is never a compile error — it follows the same missing-value convention an absent $doc section follows (resolve null, never error), and populates as soon as a BHS wrapper is present:

# Bare MSH stream (no BHS/FHS) — only the message level opens:
- msg_type: $doc.transaction_set.f08   # MSH-9 message type — populated
- batch_id: $doc.batch.f01             # null — no BHS, so no batch tier

# BHS-wrapped stream — the BHS opens a batch level, so both populate:
- msg_type: $doc.transaction_set.f08   # MSH-9 message type — populated
- batch_id: $doc.batch.f01             # BHS field — now populated

See HL7 v2 Format for the full reference.

A reader for such a format opens and closes each nested level as it crosses the corresponding envelope boundary mid-file. Each level contributes its own sections to $doc. There is no new $doc syntax for nesting — every level’s sections are read through the same two-level $doc.<section>.<field> lookup. A record inside the innermost level sees every enclosing level’s sections at once. For X12 the interchange header is a declared segment: "ISA" envelope section (you choose its name), while the GS group and ST set surface automatically as the reader-supplied sections functional_group and transaction_set, each keyed by positional eNN elements:

cxl: |
  emit interchange_control = $doc.interchange.e13        # ISA13, declared section
  emit functional_id       = $doc.functional_group.e01   # GS01 (reader-supplied)
  emit transaction_type    = $doc.transaction_set.e01     # ST01 (reader-supplied)
  emit claim_amount        = amount                       # body field

A record streamed inside the ST level resolves the ST section, the enclosing GS section, and the outermost ISA section, all at once: each inner level inherits every enclosing level’s sections as siblings in one flat namespace. If two levels declare a section with the same name, the innermost wins for records inside it — the same shadowing rule a nested scope follows in any language. Picking distinct per-level names (as above) keeps every level independently visible.

The reader-supplied default names are not the only option: an X12 source can name the GS and ST levels itself and give each a typed field schema, so a nested level is addressable under a chosen name with coerced fields exactly like the declared ISA section. The declaration lives on the source’s options (not the envelope: block, which is reserved for the pre-scannable file-level header), and each nested level is named independently:

type: x12
options:
  group_section:
    name: functional_group       # your choice — the engine reserves no name
    fields: { e06: int }         # GS06 group control number, typed
  set_section:
    name: transaction_set        # your choice
    fields: { e01: int }         # ST01 transaction-set id, typed

Omit a level’s declaration and it falls back to its reader-supplied default name keyed by untyped positional eNN strings. See X12 Format for the full reference.

Boundaries nest correctly through the pipeline: each level opens before the records inside it and closes after them, in strict innermost-first order. A level that arrives across several branches is still handled once where a Merge or Combine brings those branches together — exactly like a single-level document.

Header-only interchanges

A multi-level envelope file can legitimately carry an interchange whose body is empty — envelope structure (an interchange header, and possibly inner group headers) with zero records inside. Such an interchange still opens a document and emits its open/close boundaries, so downstream operators and trailer-count validation observe it just like any other document. The interchange’s $doc.* sections are extracted and the boundaries flow even though no body record ever streams from it.

The same holds for an empty inner envelope — an open/close pair with no records between — and for an inner envelope that opens or closes after the file’s last body record. Every envelope boundary a reader signals is applied, whether or not a record follows it, so the document frame stays balanced end to end.

Fixed-width document output

A fixed-width Sink can echo any extracted section as its header or footer. The names select document context; they do not declare new input extraction rules. For example, these Sink config keys select sections named manifest and totals:

reconstruct_envelope: true
options:
  envelope:
    header_from_doc: manifest
    footer_from_doc: totals

Both names are author-defined. With a multi-record flat-file source, both sections must have been captured in the leading header region, even though totals is rendered at the end of the output document. A structural trailer count check does not make that trailing input record an extracted section.

Header and footer fields concatenate in stored order without the body’s fixed-width padding. Strings remain verbatim, numbers and booleans use scalar spellings, dates use YYYYMMDD, datetimes use YYYYMMDDhhmmss, and null adds no text. Arrays and maps are unsupported: a structured field anywhere in the section rejects that entire header or footer before delivery. A missing section emits nothing; a present empty section emits only the configured LF/CRLF separator, or no bytes with line_separator: none.

Document start, each record, and document end are separate prepared operations. Only successful delivery advances document state or the body count; opening the next document starts its own count and selects its own sections. Earlier successful operations remain delivered if a later footer fails. A computed footer record-count field is unsupported for fixed-width (E346). See fixed-width output and output preparation.

Native JSON and XML output boundaries

JSON/XML reconstructed envelopes prepare document start, each body record, document end and finalization as separate complete operations. A document’s body count advances only when a record is delivered. Ending one document and opening the next resets that count and restores the next document’s own sections. Section names remain the names declared in your pipeline.

Reconstructed JSON envelopes retain their document grammar: array mode wraps the envelope documents in an array; NDJSON mode separates complete envelope documents with LF. Unlike ordinary compact NDJSON records, reconstructed pretty: true documents can span lines. XML places each Document wrapper inside one configured root, retaining header/footer section wrappers and emitting no XML declaration.

Explicitly opened empty documents still carry framing and a zero body count. In the CLI, a native source with no body records does not open a writer, so its output is an empty file even with reconstruction enabled. Do not infer this behavior from header-only message interchanges, whose readers emit explicit boundaries even without a body.

Ordinary native output supports correlation, splitting and per-file fan-out. Reconstructed envelopes combined with splitting, per-file fan-out, correlation or document-grain DLQ are rejected before execution (E347). These combinations cannot establish the required single document boundary.

Library writers distinguish draining already delivered bytes from finalizing a document. A byte drain adds no closing syntax. Once delivery fails, neither a drain, finalization nor teardown resumes the failed operation. See output preparation for failure effects.

Error Handling & DLQ

Clinker provides structured error handling with a dead-letter queue (DLQ) for records that fail processing. The error_handling: block at the top level of the pipeline YAML controls the behavior.

Configuration

error_handling:
  strategy: continue
  dlq:
    path: "./output/errors.csv"
    include_reason: true
    include_source_row: true

Strategies

error_handling.strategy is pipeline-wide – it is set once at the top level, not per node. It controls what happens when a record fails:

There are exactly two, because the engine makes exactly one decision at each record failure: propagate it and stop, or dead-letter it and carry on.

fail_fast

The safest strategy. Any record-level error (type coercion failure, validation error, missing required field) halts the pipeline immediately, with a non-zero exit and no DLQ file. Use this when data quality is critical and you prefer to fix issues before reprocessing.

Some failures abort the run under either strategy, because they are not record-scoped: an unwritable output path, a config or CXL compile error, and the DLQ-rate ceiling (dlq.max_rate, E315/E316) all end the run regardless of the strategy.

CSV, JSON and XML resource failures are also fatal under either strategy. Memory or disk admission refusal, allocation failure, descriptor exhaustion and temporary-storage failure are not bad-record errors, so continue cannot turn them into successful output. A typed resource diagnostic preserves the kind of failure rather than reporting every case as a memory shortage. A failed destination can already have accepted a prefix; see output preparation.

Malformed JSON/XML input encoding is a data failure, including when discovered during schema discovery or envelope pre-scan. Under fail_fast, the CLI returns exit 4 and machine code source.data.invalid. It does not report a compilation error merely because no record has reached the pipeline. A late error can leave an already delivered prefix; an envelope pre-scan may discover it before any body records. Failed runs do not publish their staged normal output files.

Explicit cancellation ends an interrupted run with exit 130; it does not add a Sink error. If a real I/O or resource failure occurs alongside a shutdown request, the real failure retains its classification. Record and byte counters describe established progress, not rows merely attempted or prepared.

An executor invariant failure also aborts under either strategy with exit code 1. In particular, if a planned materialized input is unavailable when its consumer runs, Clinker stops instead of treating that input as a legitimate zero-row result. The message names the consuming node and planned producer (including the producer port when applicable) and says the input was not treated as empty. A source or stage that really emits zero rows remains valid; it carries an explicit empty buffer and completes normally. Report any missing- input internal error as an engine defect rather than routing it to the DLQ.

continue

The production workhorse. Bad records are written to the DLQ file with diagnostic metadata, and the pipeline continues processing remaining records. After the run completes, inspect the DLQ to understand and correct failures.

A pipeline that completes with DLQ entries exits with code 2 – this signals “pipeline completed successfully but some records were rejected.” It is not a crash or internal error. A continue run that dead-letters nothing exits 0, exactly like a clean fail_fast run.

Migrating from best_effort. The removed best_effort spelling was a third name for the continue behavior: it wrote the same DLQ entries and produced the same exit code, because the runtime never distinguished the two. Replace it with strategy: continue. A pipeline still carrying best_effort is rejected at config-validation time with a message naming the replacement.

Declared source-type failures are deliberately not lossy: under continue, the complete original record is written to the configured DLQ and no null, raw, or partially converted replacement enters the pipeline. This strategy therefore requires an error_handling.dlq block when such a failure occurs. fail_fast stops on the first failure without emitting a replacement. This includes fields renamed by a source schema: rejection retains the original decoded record and its values, even when conversion failed after other fields had already been examined.

An evaluation error is never false

A condition that fails to evaluate, such as one that divides by zero, has not said whether it holds. The engine never reads that failure as “false”, on any node: the record the condition was about is dead-lettered, and no decision that depends on the condition being false is taken for it.

• A Transform filter that fails dead-letters the record; it is neither kept nor filtered out.
• A Route branch condition that fails dead-letters the record, which takes no branch and not the default. In exclusive mode only the conditions up to the first true one are evaluated, so a later condition cannot fail.
• A Combine where: that fails for a candidate build row dead-letters that pair. The driver is not unmatched, so on_miss does not fire; under match: first a failure on the deciding candidate is the driver’s only result, and under match: collect the driver writes no row. See Combine.

A pipeline that wants a failing condition treated as false says so in CXL, for example by guarding the division or coalescing the result with ?? false.

DLQ configuration

The DLQ is always written as CSV, regardless of the pipeline’s input/output formats.

  dlq:
    path: "./output/errors.csv"
    include_reason: true
    include_source_row: true

Per-source DLQ settings

per_source gives a Source its own DLQ file, its own rate ceiling, or both. Each key is the name of a Source node:

error_handling:
  strategy: continue
  dlq:
    path: ./output/errors.csv
    max_rate: 0.05
    per_source:
      vendor_feed:
        path: ./output/vendor_feed_errors.csv
        max_rate: 0.20
        min_records: 500
      orders:
        max_rate: 0.01

A Source’s own max_rate is checked first, so a breach names that Source. The pipeline-wide max_rate, when set, still applies to the run as a whole. In the example, vendor_feed may dead-letter up to 20% of its own rows, but the run still stops when all dead letters together reach 5% of all rows read.

A key that does not name a declared Source is rejected at compile time (E317). Two DLQ paths that name one file are rejected too (E318), including paths that differ only in case on a case-insensitive filesystem, or in being written relatively and absolutely. A DLQ path that names the same file as a Sink’s path is rejected with E322.

Which record columns each file carries follows from the Sources routed to it; see How the DLQ columns are chosen.

How DLQ output is written

Dead-letter rows are written while the run executes, not collected until it ends. Each row is formatted under its file’s header, which is fixed when the pipeline compiles (see How the DLQ columns are chosen), and written into a staged copy of that file in the run’s publication attempt: in quarantine next to the destination by default, or under local_spool_dir with mode = "local_then_publish" (see Output publication). Each open DLQ file writes through one fixed 64 KiB buffer, so the memory the DLQ files use does not grow with the number of failures.

• A DLQ file is created when its first row arrives. A file no row reaches is not created, and no empty file is published.

• DLQ files are published only if the run succeeds, by the same publication step as the pipeline’s other outputs. A failed or interrupted run publishes no DLQ file.

• The failures that are counted but have no destination (see path above) are never formatted or written.

• Three kinds of dead letter are held in memory until the stage that found them finishes, and written then:

  • a join_values collision at a Sink that writes on its own thread;
  • an Aggregate add_record failure found while the Aggregate reads its input on its own thread;
  • a Combine output-row failure found while the Combine streams its driver on its own thread, or inside a grace-hash, sort-merge or IEJoin join.

  A Sink that writes on its own thread stops the run with an internal error once 65,536 collisions are waiting this way.

• Under a correlation key or dlq_granularity: document, records are held until their group or document is decided. That is those features’ own state, described in their sections, not DLQ output; the rows they dead-letter are then written like any other. Under dlq_granularity: document the engine also keeps, for each rejected document, a compressed record of which rows it has already written, so a row held by several Sinks is written once. That record is charged to the memory budget. While a Sink writes a rejected document’s rows, the record grows by:

  • up to 16 bytes per row when the Sink receives the rows in the order their Source read them;
  • up to 96 bytes per row when it receives them in any other order, for example after a Sort on a data column;
  • up to 672 bytes per row for a document whose rows span its Source’s 4,294,967,296th row, in any order. A Source counts its rows across every file it reads.

  The first row of each document a Sink starts costs up to 920 bytes. The growth is released when the Sink’s pass ends, or every 65,536 rows of a document. Between releases one document’s record therefore grows by at most about 1 MiB when its rows arrive in the order they were read, about 6 MiB in any other order, and about 42 MiB when its rows span the 4,294,967,296th row. For example, 65,535 rows that alternate between a document’s first 65,536 rows and its next 65,536 grow the record to about 5.8 MB, and the release at the next row brings it back to 880 bytes. Once released, the record costs at most 4 bytes per written row, plus 96 bytes for every 65,536 rows of the document, written or not, plus under 1 KiB. That is about 2.3 KiB for a million-row document whose rows are contiguous. When only one row in each 65,536 is written, it is 88 bytes per written row plus 672 bytes, within the cap of about 100 bytes per written row. The record cannot spill: if one more row would not fit once every held row (below) has moved to disk, the run fails with E310. A failed document’s failing records are formatted as DLQ rows when they fail and held until the document is rejected. They are held in memory, charged to the memory budget, and move to one file in the spill directory only when the budget needs the memory, counting toward storage.spill.disk_cap_bytes (E320). With memory to spare nothing is written to disk. If even one more held row would not fit with every held row on disk, the run fails with E310.

Disk bounds how much DLQ output a run can produce: the free space at the staging location, and the publication attempt’s byte ceiling (storage.publication.max_attempt_bytes, and no more than retained_byte_limit), which every staged file of the run counts toward, DLQ files included. An attempt larger than that ceiling is refused at publication and nothing is published.

When the staging location fills, the run fails with an I/O error (exit code 4) and publishes nothing. After the destination’s own error text, the message names the DLQ file, the number of rows dead-lettered so far, and the stage and category with the most rows, and suggests a breaker:

<destination error>: dead-letter output ./output/errors.csv could not be written after 1048576 dead-lettered rows (most from stage transform:validate_orders, category validation_failure)

help: stop the run before dead letters fill the destination, for example:

  error_handling:
    type_error_threshold: 0.05
    dlq:
      max_rate: 0.05

These breakers bound how many rows can dead-letter; disk at the staging destination bounds their volume.

Bounding how much can dead-letter

Disk bounds the volume of dead-letter output; the breakers bound how many rows can dead-letter in the first place. Set one wherever a wrong schema could make most rows fail, for example when a feed can change its columns or types without notice:

error_handling:
  strategy: continue
  type_error_threshold: 0.05
  dlq:
    path: ./output/errors.csv
    max_rate: 0.05

• dlq.max_rate (E315) and dlq.per_source.<name>.max_rate (E316) stop the run when the fraction of dead-lettered rows crosses the ceiling, once min_records rows have been read. The numerator counts dead-letter rows, collateral rows included, whether or not they have a DLQ file to go to. A source row counts once for each failure it took part in, so a row that failed twice counts twice, with or without a correlation key. Under dlq_granularity: document, a row that failed twice at the Source, in a Transform or in a Route is written once and counts once (see Document-level DLQ).
• type_error_threshold (E368) stops the run when the fraction of declared source-type failures crosses the threshold. It catches a schema mismatch at the Source, before the failing rows reach later stages.

A rate ceiling is checked each time a dead letter is counted, so the row that crosses it is itself counted and written before the run stops. A stopped run exits with code 3 and publishes nothing. No breaker is set by default.

DLQ columns

Every DLQ record includes these metadata columns:

Timestamps need not increase down a file. A failure can be held before its row is written, for example in a correlation group that commits later, or by a stage listed under How DLQ output is written, so its row can follow rows observed after it. Sort on _cxl_dlq_timestamp or _cxl_dlq_id to read failures in the order they were observed.

When include_reason: true is set, two additional columns appear:

Pairing the rows one failure produced

Two rules decide which rows a DLQ file holds:

• One row per failure. Every failure writes its own trigger row, with its own category, detail, triggering field and value, stage and route. A row that fails twice, on two Route branches or against two Combine build rows, is written twice. A Combine failure also writes the build row that contributed to it, right after its trigger. Under dlq_granularity: document this rule has an exception: a rejected document’s later failing record is written as a document_rejected row paired with the document’s first failure, and a record that failed twice at the Source, in a Transform or in a Route is written once (#1316; see Document-level DLQ).
• Each condemned row once. A correlation group or document that fails adds each of its other rows once, as a collateral of its first failure. Under dlq_granularity: document, a record that an Aggregate, a Combine or a Reshape failure dead-lettered is written again, as a document_rejected row, only when a Sink on another branch also received it and its document is rejected. If its document is not rejected, that Sink can publish it (see Not covered).

A correlation key never removes, merges or relabels a failure row: the same failures are written with and without a key, and the key only adds the rows a failing group condemns and decides when rows are written.

One failure can dead-letter several rows:

• a Combine body that fails writes the driver row and its matched build row, each under its own Source’s _cxl_dlq_source_name and _cxl_dlq_source_row, whichever join strategy ran. The build row is written once per failure, so a build record that two failing drivers matched is written twice, once after each driver’s row, and a driver that fails against three build rows is written three times, each copy followed by its own build row;
• a failing row in a correlation group takes the rest of its group with it as correlated rows;
• a group larger than max_group_buffer writes a group_size_exceeded row and the rest of its group as correlated rows (the group’s own failures keep their own trigger rows, see below);
• under dlq_granularity: document, a failing record rejects the rest of its document as document_rejected rows.

Every row carries _cxl_dlq_trigger_id, the _cxl_dlq_id of the trigger row whose failure produced it, and that trigger row is always in the run’s output. A trigger row points to itself, so _cxl_dlq_trigger is true exactly when _cxl_dlq_trigger_id equals _cxl_dlq_id. Group by _cxl_dlq_trigger_id to see everything one failure took with it. In this excerpt (other columns omitted), the first two rows are a Combine driver row and its build row, and the last two are a correlation trigger and one of its collaterals:

_cxl_dlq_id,_cxl_dlq_trigger_id,_cxl_dlq_source_name,_cxl_dlq_error_category,_cxl_dlq_trigger
01928f3a-6c10-7b21-8a4e-3f1c2d9e0a01,01928f3a-6c10-7b21-8a4e-3f1c2d9e0a01,orders,combine_output_row,true
01928f3a-6c10-7b22-9f07-51e6a8b4c302,01928f3a-6c10-7b21-8a4e-3f1c2d9e0a01,rates,combine_output_row,false
01928f3a-6c14-7c03-b2d8-0a9e7f615203,01928f3a-6c14-7c03-b2d8-0a9e7f615203,employees,type_coercion_failure,true
01928f3a-6c19-7d40-8c11-6e2b90d3f404,01928f3a-6c14-7c03-b2d8-0a9e7f615203,employees,correlated,false

A trigger whose failure wrote nothing else carries its own id and shares it with no other row. Every row has its own _cxl_dlq_id, so the value only repeats across the rows of one failure.

When a correlation group holds several failing rows, each failing row is a trigger and keeps its own id as its trigger id. The group’s correlated rows carry the trigger id of the group’s first failing row, the one whose error their _cxl_dlq_error_detail quotes. A rejected document has one trigger, its first failing record; its other records, including any that failed after it, carry that trigger’s id.

A group larger than max_group_buffer can hold failing rows too. Each of them is still written as its own trigger, with its own category and id, and before the rest of the group. The group’s other rows follow under one group_size_exceeded trigger, the first of them, and the remaining ones are correlated rows carrying its id. Each failure is written once, as its own trigger; a row that failed on one Route branch and reached a Sink on another is written as its own failure, and never again as correlated or as the overflow trigger. A group whose rows all failed writes no group_size_exceeded row, because the overflow took nothing with it that had not already failed.

The rows of one failure can land in different DLQ files: with per_source paths, a Combine build row goes to its own Source’s file while its driver row goes to the driver’s.

How the DLQ columns are chosen

Each DLQ file’s header is fixed when the pipeline compiles, before any record is read. It does not depend on which records failed, or on which stages they failed in: every time a pipeline writes a given DLQ file, that file has the same columns in the same order.

A header starts with the _cxl_dlq_* metadata columns, always in this order: _cxl_dlq_id, _cxl_dlq_trigger_id, _cxl_dlq_timestamp, _cxl_dlq_source_file, _cxl_dlq_source_name, _cxl_dlq_source_row, _cxl_dlq_triggering_field, _cxl_dlq_triggering_value, then _cxl_dlq_error_category and _cxl_dlq_error_detail when include_reason is on, then _cxl_dlq_stage, _cxl_dlq_route and _cxl_dlq_trigger.

With include_source_row on, the record columns follow. They come from every record shape that can reach that file:

• the declared columns of each Source, and _cxl_dlq_source_record, when Source rejections are dead-lettered (strategy: continue);
• the shape of the records entering each Transform, Route, Reshape, Aggregate, Combine and Sink that can dead-letter a record under the pipeline’s strategy (stages inside a composition count at the composition’s place in the pipeline);
• the output columns of an Aggregate without group_by under strategy: continue, which go to the pipeline-wide file.

A shape is added to the file of each Source whose records can carry it: that Source’s per_source.<name>.path when it has one, otherwise the pipeline-wide path. A shape that carries no Source identity, such as the output of a Combine using match: first or match: all, is added to the pipeline-wide file.

The shapes then combine as follows:

• One shape keeps its natural column order.
• Several shapes give their first-seen union in plan order: each column appears once, at the position where it was first seen. Plan order is the node order --explain prints, which need not match the order the nodes are written in the YAML.
• A column a record does not carry is an empty cell in that record’s row.
• Engine-only sidecar columns ($widened and the $source.* stamps) are never written. Correlation-key columns ($ck.*) are.

For example, this pipeline has two Sources with different columns, one pipeline-wide DLQ file, and a Transform on each Source that can fail:

error_handling:
  strategy: continue
  dlq:
    path: rejects.csv
nodes:
- type: source
  name: orders
  config:
    schema:
      - { name: order_id, type: int }
      - { name: amount, type: int }
    # ...
- type: source
  name: refunds
  config:
    schema:
      - { name: refund_id, type: int }
      - { name: order_id, type: int }
      - { name: amount, type: int }
    # ...
# one Transform on each Source, then a Sink on each Transform

In this plan refunds comes before orders, so the record columns of rejects.csv are:

refund_id,order_id,amount,_cxl_dlq_source_record

An orders row writes an empty refund_id cell. order_id and amount appear once, although both Sources declare them. _cxl_dlq_source_record is in the header because a Source rejection can reach the file under continue. It is empty for these rows.

To see every DLQ file’s columns before a run, use clinker run pipeline.yaml --explain: its === Dead-Letter Output === section lists each file, the Sources routed to it and its full header (see Explain Plans).

Source-file provenance (_cxl_dlq_source_file) is read from each record, so one file’s path is never reused for a later row.

Error categories

The _cxl_dlq_error_category column contains one of these values:

Advanced options

Type error threshold

Abort the pipeline if the fraction of declared source-type failures exceeds a threshold:

  type_error_threshold: 0.05    # Abort if >5% of records fail

The cumulative ratio is:

declared source-type failures / decoded source rows observed

The rejected row appears once in both numerator and denominator. The same typed-error event is used for strategy routing, DLQ accounting, and this circuit breaker; unrelated validation, structural-document, and collateral DLQ entries do not enter the numerator. Equality is allowed: a threshold of 0.05 stops only when the ratio is strictly greater than 5%. 0.0 stops on the first type failure, while 1.0 never trips. Values must be finite and in [0.0, 1.0].

Correlation key

Declare correlation_key on the contributing Source’s config: block, not on error_handling:. Group DLQ rejections by a key field. When any record in a correlation group fails, records from the failing source’s contribution to that group are routed to the DLQ:

# Inside a Source's config:
correlation_key: order_id

For compound keys:

# Inside a Source's config:
correlation_key: [order_id, customer_id]

This is useful for transactional data where partial processing of a group is worse than rejecting the entire group. For example, if one line item in an order fails validation, you may want to reject the entire order.

Under multi-source ingest, the collateral fan-out narrows to the failing source: a src_b trigger does NOT DLQ records from src_a that share the same correlation key. Single-source pipelines see bit-identical behavior to today’s pipeline-wide collateral DLQ. See Per-source rollback narrowing for the full semantic and the two documented exceptions (max_group_buffer overflow and Combine output failures).

When a Combine output row fails under a correlation key, the failing driver row is the trigger of the driver’s correlation group. The dead letter for the matched build record is held with that group as a collateral (_cxl_dlq_trigger: false, category combine_output_row), written right after its driver’s row with its driver’s _cxl_dlq_trigger_id, and written or rolled back exactly when that group is. It never condemns the build record’s own correlation group, so another driver that matched the same build record keeps its output unless its own group failed. Each failure gets its own copy of the build row, paired with its own driver row: two failing drivers of one group each get one, and a driver that fails against several build rows is written once per failure, each copy followed by the build row of that failure. The held failure keeps its triggering field and value, as it does without a key.

Because every failure is written, a correlation key does not lower dlq_count, records_dlq or the max_rate numerators: they equal the counts the same failures give without a key, plus the rows the failing groups condemn.

For the full lifecycle and per-operator semantics (route, merge, aggregate, combine), see Correlation Keys.

Max group buffer

Limit the number of records buffered per correlation group:

  max_group_buffer: 100000     # Default: 100,000

A group that goes over this limit is dead-lettered whole. Its failing rows are written as their own triggers; its other rows are written under one group_size_exceeded row, as correlated rows. The limit counts held entries, not distinct rows: see Correlation Keys.

Document-level DLQ

By default a record failure dead-letters only that record (dlq_granularity: record). A source can instead reject the entire document any record of which fails, by declaring the granularity per source:

nodes:
  - type: source
    name: claims
    config:
      name: claims
      type: x12
      glob: ./claims/*.edi
      schema: [{ name: seg_id, type: string }]
      dlq_granularity: document   # record (default) | document

Under dlq_granularity: document and the continue strategy, a document is rejected when one of its records fails at the Source (against its declared type, or a structural rule), in a Transform, or in a Route. A record that fails in an Aggregate, a Combine or a Reshape is dead-lettered as its own row, as under record granularity, and does not reject its document (#1232; see Not covered). When a document is rejected:

• the failing record becomes the root-cause DLQ entry (_cxl_dlq_trigger = true, carrying its original error category);
• every other record of the document that reaches a Sink becomes a collateral entry (_cxl_dlq_trigger = false, category document_rejected); a record dropped before any Sink is not written;
• every other record of the document that fails at the Source, in a Transform or in a Route is also a document_rejected collateral, written right after the root cause in the order the records failed. These records are stamped (_cxl_dlq_id, _cxl_dlq_timestamp) when they fail; the records a Sink held are stamped when the Sink rejects the document. Every collateral carries the root cause’s id in _cxl_dlq_trigger_id;
• no record read from that document is written by any Sink, however many Sinks read it. Rows derived from the document’s records can still reach a Sink; see Not covered.

Clean documents in the same run stream through untouched, and records from sibling sources still on the default record granularity keep per-record semantics — the policy is per source.

Sinks run last. Every Sink runs after every other node, so a document’s verdict is final before any Sink writes one of its records. --explain lists the Sinks last. Dead-letter rows that other nodes write directly come before the rows a Sink writes.

Several Sinks. Each source row of a rejected document appears in the DLQ once, however many Sinks reached it. A row that failed is written as it was when it failed, ahead of any Sink’s copy of it. Any other row is written as it was held by the first Sink, in run order, that held it; a row that reached only a later Sink (a Route sent it there, say) is written by that Sink. Rows are matched by source row, so of the records one emit each makes from a source row, one is written. Every collateral, whichever Sink writes it, names the document’s root-cause entry in _cxl_dlq_trigger_id.

There is one exception, described under Not covered: a record that an Aggregate, a Combine or a Reshape failure dead-lettered (for a Combine, the driver row and the build row it matched) is written for that failure. It is written again, as a document_rejected row, only when a Sink on another branch also received it and its document is rejected; if its document is not rejected, that Sink can publish it (#1232). The rows a Combine or an Aggregate writes for a rejected document, and the rows that pass through a Reshape, are not held back at all.

This is the document-shaped analogue of correlation keys: use it when partial processing of a document (an EDI interchange, a batch file with a header/trailer) is worse than rejecting the whole document. Unlike correlation keys, which group across files by a key value, document-level DLQ scopes rejection to a single document’s records.

Document grain. The document is the outermost level — the source file. For a flat format (CSV, JSON, plain XML) each input file is one document. For a nested-envelope format (an X12 ISA → GS → ST interchange, an EDIFACT UNB → UNG → UNH) the document is the whole interchange / file, not an inner functional group or transaction set: a failure anywhere in the interchange rejects the entire interchange, including the transaction sets that validated cleanly. Reject the inner-level grain instead by partitioning the input so each interchange is its own file is not currently offered — the grain is fixed at the file.

DLQ rate. Each dead-letter row — the trigger and each collateral — counts once toward the configured DLQ max_rate, matching the correlated-collateral precedent, however many Sinks held it; a record written twice under the exception in Several Sinks counts twice. A rejected 1000-record document contributes 1000 when every one of its records reaches a Sink. It does not affect type_error_threshold, whose numerator contains only declared source-type failures.

dlq_count never counts a row that ok_count also counts, because no Sink writes a record read from a rejected document, with these exceptions, each described under Not covered:

• a record that an Aggregate, a Combine or a Reshape failure dead-lettered, which a Sink on another branch can still write when its document is not rejected (#1232);
• a rejected document’s rows that a Combine or an Aggregate writes, or that pass through a Reshape, which reach a Sink while the document’s records are dead-lettered;
• a record whose joined or aggregated row fails in a node after the Combine or Aggregate: that failure is written while the record itself can still reach a Sink on another branch.

Memory. The engine buffers each open document’s records until its boundary, then flushes the document clean to the sink or rejects it and drops the buffer. Peak memory scales with the concurrently-open documents, not the total input; a single very large document spills its buffer to disk under the run’s memory budget rather than holding everything in RAM. See Streaming vs blocking for the spill model.

Sink restriction. Document-level DLQ flushes each whole document to a single output writer, so it cannot be combined with a per-source-file Sink (a {source_file} / {source_path} path template over a multi-file source). The two are rejected together at compile time (E343); use a single output path, or set dlq_granularity: record if per-file output is the requirement.

Strategy requirement. dlq_granularity: document requires error_handling.strategy: continue. It is incompatible with the default fail_fast: document-level dead-lettering keeps the run going past a bad document, which contradicts fail-fast’s abort-on-first-error. The combination is rejected at compile time (E344) — set strategy: continue to dead-letter bad documents, or keep fail_fast with the default dlq_granularity: record.

Correlation restriction. Document-level and correlation-key rejection are alternative atomic-disposition models. A document is keyed by its source file; a correlation group can span files and is keyed by authored field values. The engine does not define precedence or a combined writer boundary for those two populations, so a pipeline containing both dlq_granularity: document and any correlation_key is rejected at compile time (E370). Remove every correlation_key to keep document rejection, or set dlq_granularity: record to keep correlation rejection.

Composition restriction. A composition body may not declare a Sink when any Source uses dlq_granularity: document (E378): a body Sink runs inside its composition, where it cannot be held back until every document’s verdict is final. Declare the Sink at pipeline level instead. Where moving it through a new composition output port works today, the diagnostic prints that move ready to paste. Otherwise it names clinker explain --code E378, which shows how to declare the Sink’s work at pipeline level.

Not covered. Document rejection does not yet reach these rows:

• A row failure inside an Aggregate, a Combine or a Reshape is dead-lettered per record and does not reject its document (#1232). When a Sink on another branch also received that record, it is written again as a document_rejected row if another failure rejects its document, and that Sink can publish it if none does.
• The rows a Combine or an Aggregate writes are not held back by their document’s verdict, so a rejected document’s joined or aggregated rows still reach a Sink. A failure on such a row, in any node after the Combine or Aggregate, dead-letters only that row and does not reject its document, so the document’s other records, and that row’s own source record on another branch, can still be published (#1317).
• A Reshape’s output rows do not keep their input row’s document, so a rejected document’s rows that pass through a Reshape still reach a Sink.
• In a pipeline where any Source declares dlq_granularity: document, a CSV join_values collision with on_conflict: error at a Sink fails the run rather than dead-lettering the record (#933).

A document is identified by its file path, so two Sources reading the same file share one verdict.

Spilling stages. Document identity survives memory pressure end to end. The per-document buffer identifies each document before buffering and spills under the memory budget, and a Sort between the source and the output preserves each record’s document context — including the source file the grain keys on — across its own spill round-trip. A document whose records pass through a spilling Sort is therefore still grouped and rejected as one document under memory pressure, exactly as it would be in memory. The rows a hash Aggregate or a grace-hash Combine writes are not held back by their document’s verdict, spilled or not; see Not covered.

Malformed envelopes (structural validation)

Envelope formats carry their own structural-integrity claims: an X12 interchange declares a segment count in each SE/GE/IEA trailer, EDIFACT in each UNT/UNZ, HL7 batch/file in each BTS/FTS, and a multi-record flat file’s trailer record declares a body count via its structure: constraint. When the declared count does not match the body the reader actually streamed, the file is structurally invalid. A multi-record flat file can also break a non-count structural rule — a line whose record-type discriminator matches no declared records: entry (E345), or a body record appearing after the trailer that closes the document; these are classified separately from a count mismatch but carry the same disposition.

Under dlq_granularity: document, such a structural failure dead-letters the whole source file to the DLQ rather than aborting the run:

• the file’s records dead-letter as one structural_validation root-cause entry (_cxl_dlq_trigger = true) plus a document_rejected collateral for every other already-streamed record of the file;
• no record of the malformed file reaches the success sink.

nodes:
  - type: source
    name: claims
    config:
      name: claims
      type: x12
      glob: ./claims/*.edi
      schema: [{ name: seg_id, type: string }]
      dlq_granularity: document   # reuse the document opt-in — no separate config

The opt-in is the same dlq_granularity: document knob that governs per-record document rejection above; there is no separate validation: block. A malformed envelope is simply one more reason a source under the document policy condemns a whole document. E345 is the one structural class that also has a record-grained recovery: under strategy: continue and the default dlq_granularity: record, only the unknown-tag row is dead-lettered and the reader continues at the next physical row. The DLQ row exposes the unknown tag as record_type and the unguessed decoded input as _cxl_dlq_source_record (fixed-width line text, or a JSON array of decoded CSV cells).

Honest timing — rejected at the sink boundary, not before the first record. The trailer that carries the count arrives at the end of the file, after every body record it counts has already streamed through the DAG. Clinker is a bounded-memory streaming engine — it does not buffer the whole file up front to pre-validate it (that would defeat the streaming model). So the count mismatch is detected mid-stream, the file is marked failed, and the document-level DLQ buffer rejects every already-streamed record of the file at its close. The user-visible outcome is the same — no record of a malformed envelope is ever written to the output — but the rejection lands at the sink boundary, not literally before the file’s first record streams.

Grain — the whole file. An SE-level mismatch (one transaction set inside a larger interchange) rejects the entire interchange / file, not just that one transaction set, because the document grain is the outermost source file (see Document grain above). Split the input so each interchange is its own file if you need finer rejection.

Multiple files keep flowing. When a glob / paths source matches several files and one is malformed, only that file dead-letters — ingestion continues to the remaining files, so the clean files after a bad one still reach the sink. (This is unlike the default record granularity, where a count mismatch aborts the whole run and no file’s records are written.) Dead-lettering one malformed file never silently drops the good files around it.

Record-grained E345 is narrow. Under the default dlq_granularity: record, strategy: continue can recover only from an unknown multi-record discriminator because the reader has consumed exactly one bounded physical row and can resume unambiguously. Trailer-count mismatches and a body record after a document-closing trailer still abort at record granularity: neither belongs to one independently recoverable row. Genuine corruption (a truncated stream, a bad delimiter, a control-number echo mismatch, a segment after an X12/EDIFACT/HL7 envelope trailer) always aborts, even under the document opt-in. Under fail_fast, E345 also aborts at the offending line.

Cryptographic integrity (checksums / signatures) is not yet validated. Envelope formats can also carry a SHA-256 body hash, a JWS-signed JSON payload, or an XML Signature. Clinker extracts these envelope sections but does not yet verify them. Tracked for a future release.

Exit codes

Exit code 2 is not a failure – it means the pipeline ran to completion and handled errors according to the configured strategy. Check the DLQ file for details. See Exit Codes & Error Diagnosis for the full reference and the orchestrator retry policy.

Complete example

pipeline:
  name: order_processing
  memory: { limit: "512M" }

nodes:
  - type: source
    name: orders
    config:
      name: orders
      type: csv
      path: "./data/orders.csv"
      correlation_key: order_id
      schema:
        - { name: order_id, type: int }
        - { name: customer_id, type: int }
        - { name: amount, type: float }
        - { name: email, type: string }

  - type: transform
    name: validate_orders
    input: orders
    config:
      cxl: |
        emit order_id = order_id
        emit customer_id = customer_id
        emit amount = amount
        emit email = email
      validations:
        - field: email
          check: "not_empty"
          severity: error
          message: "Customer email is required"
        - check: "amount > 0"
          severity: error
          message: "Order amount must be positive"

  - type: sink
    name: valid_orders
    input: validate_orders
    config:
      name: valid_orders
      type: csv
      path: "./output/valid_orders.csv"

error_handling:
  strategy: continue
  dlq:
    path: "./output/rejected_orders.csv"
    include_reason: true
    include_source_row: true
  type_error_threshold: 0.10

Nodes

A Clinker pipeline is a single flat nodes: list. Every entry carries a type: discriminator that selects a node kind — the unified node taxonomy. There is no separate “join section” or “filter section”: records flow through one homogeneous graph of typed nodes, wired together by input: / inputs:. This part documents the ten runtime node kinds; an eleventh, Composition, is a call-site that inlines a reusable sub-pipeline and is covered under Pipelines.

The pages in this part are ordered the way data flows through a DAG — a record enters at a Source, fans through the record-level and combining nodes, and leaves at a Sink:

Streaming vs blocking

Stateless nodes (Transform, Route, Merge, the Combine probe side, Sink) evaluate records one at a time without accumulating per-record state. Blocking nodes (Aggregate, sort, the grace-hash Combine build side) accumulate state inside the RSS budget and spill to disk rather than OOM the process. The Streaming vs. Blocking Stages page in the Operations Guide is the full memory model.

Wiring and naming

Every node needs a unique name: (no dots — the dot is reserved for port syntax). Single-input nodes use input:; Merge and Combine use inputs:. Route branches are consumed downstream as route_name.port. The Pipeline YAML Structure page covers the full wiring grammar, optional fields (description:, _notes:), and strict parsing rules.

Source Nodes

Source nodes read data from files and are the entry points of every pipeline. They have no input: field – they produce records, they do not consume them.

Basic structure

- type: source
  name: customers
  config:
    name: customers
    type: csv
    path: "./data/customers.csv"
    schema:
      - { name: customer_id, type: int }
      - { name: name, type: string }
      - { name: email, type: string }
      - { name: status, type: string }
      - { name: amount, type: float }

Schema declaration

The schema: field is required on every source node. Runtime ingestion does not guess types from data: declare each column’s name and CXL type explicitly. This schema drives compile-time type checking across the entire pipeline.

Each entry is a { name, type } pair:

schema:
  - { name: employee_id, type: string }
  - { name: salary, type: int }
  - { name: hired_at, type: date_time }
  - { name: is_active, type: bool }
  - { name: notes, type: { nullable: string } }

Available types

Check the schema through the planner

The schema that governs execution is the one accepted by clinker-plan while compiling the complete pipeline. Run this after changing a source schema:

clinker run pipeline.yaml --explain text

Exit code 0 means the planner parsed the canonical YAML, resolved schema references and overlays, bound the pipeline, type-checked CXL, and produced a compiled plan. It does not prove that input bytes satisfy the declarations or that the resulting output is correct; execute representative input against an isolated destination and compare the result before a production run.

Workspace tooling may also present advisory schema analysis. That analysis can help find likely authoring mistakes, but it cannot admit or reject a pipeline. Its analyzed, partial, skipped, and failed statuses describe only how much the advisory model inspected. See Validation and Admission for the status meanings and known limits.

Without a column format:, date_time accepts the default offset-free forms and RFC 3339 timestamps with Z or an explicit numeric offset, such as 2026-01-31T08:27:00Z and 2026-01-31T10:57:00+02:30. Zoned timestamps are normalized to UTC before entering Clinker’s timezone-free date_time representation. Parsing is exact: surrounding whitespace, malformed calendar values, and out-of-range timestamps are rejected rather than trimmed, guessed, overflowed, or rounded. A column-level strftime format: is exclusive: only that authored format is tried.

A source column’s declared type must be concrete: numeric — the inference-only int | float union CXL resolves during type unification — is not a valid source column type. Declaring one is rejected at compile with E158; declare int or float explicitly.

clinker guess pipeline.yaml provides an authoring-only preview for columns explicitly marked numeric. It proposes concrete int or float declarations; it does not infer arbitrary schemas or change the runtime admission rules. The default preview is bounded and read-only. --check exhausts the admitted manifest; --write can publish one guarded, unambiguous inline edit. Resolve all numeric declarations to concrete types before executing the pipeline. See Numeric schema authoring and the guess command.

long_unique — storage hint for high-cardinality text

A string column may carry an optional long_unique: true flag. It is an advisory, opt-in hint, not a type change: it tells Clinker the column’s values are long and effectively unique — never repeated across records — so the run uses less memory for that column. Typical candidates are UUIDs rendered as text, street addresses, and free-text comment or note fields.

schema:
  - { name: ticket_id,  type: string, long_unique: true }   # 36-char UUID
  - { name: notes,      type: string, long_unique: true }   # free text
  - { name: department, type: string }                      # low-cardinality, default

The flag lowers memory use only. A value’s content and its comparison, grouping, join, sort, and output behavior are all unchanged — a long_unique value behaves identically to the same text in any other column. Omitting the flag (the common case) leaves the default behavior untouched. Set it only when you know a column is genuinely high-cardinality free text; on a column whose values repeat, leave it off.

source_name — read a differently-named physical column

A column may carry an optional source_name naming the physical input column it reads from, when that differs from the exposed name. The reader matches input fields by physical name and re-labels the value under name, so downstream CXL and the output see name carrying the physical column’s data.

schema:
  # read the physical `cust_id` column, expose it downstream as `customer_id`
  - { name: customer_id, type: string, source_name: cust_id }

Omitting source_name (the common case) reads the input field whose key equals name, unchanged from before. A channel schema patch’s rename op sets this alias automatically (see Channels).

Transport vs format

A source declaration has two independent layers:

• Transport (transport:) selects where the records come from. Two transports exist: file — read bytes from the filesystem, resolved through one of the file matchers (path / glob / regex / paths) — and rest — pull records from a paginated HTTP endpoint under a hard page/record cap (see Network Sources (REST)). transport: is optional and defaults to file, so a source that omits it reads from disk exactly as before. rest needs the rest capability, which the released binary has; a build compiled without it refuses a rest source at validation with E223 rather than running the pipeline short one input (see Optional capabilities).
• Format (type:) selects how the bytes decode into records: csv, json, xml, fixed_width, edifact, x12, hl7, swift.

- type: source
  name: orders
  config:
    name: orders
    transport: file        # optional; this is the default
    type: csv              # the on-disk format
    path: "./data/orders.csv"
    schema:
      - { name: order_id, type: int }

A file transport requires exactly one file matcher (path, glob, regex, or paths). Declaring none fails validation with E211; declaring more than one fails with E210. Both are reported at config-load time, before any file is opened.

Choosing files

Interactive companion: the file discovery explainer runs these steps on a sample folder as you change the settings.

Before reading, a file Source builds its list of files in a fixed order:

1. Matcher. Paths are relative to the pipeline file’s folder.
   • path: names one file and paths: a list. A file that does not exist stops the run with E216, whatever on_no_match says.
   • glob: matches names inside one folder: * does not cross a /. Use ** to include subfolders (./data/**/*.csv). A leading dot is not special, so *.csv also matches .draft.csv. An invalid pattern is E212.
   • regex: searches the pipeline’s folder and, by default, every subfolder, and matches anywhere in each file’s path. That path begins with the folder part of the pipeline path as you typed it (clinker run pipelines/orders.yaml gives pipelines/data/orders_2024-01.csv), so anchor the end of the path, for example data/orders_\d{4}-\d{2}\.csv$. An unanchored pattern can pick up earlier outputs too. An invalid pattern is E213.
2. exclude: – a list of glob patterns; a file whose name or full path matches any of them is dropped.
3. Anything that is not a regular file is dropped.
4. min_size: / max_size: – decimal units: 1KB is 1000 bytes, also B, MB, GB.
5. modified_after: / modified_before: – a duration back from the time the pipeline is loaded (30s, 15m, 2h, 3d) or an RFC 3339 timestamp (2024-03-01T00:00:00Z).
6. files.sort_by: name (default; the full path), created or modified, with files.sort_order: asc (default) or desc. The files of a paths: list are sorted too: the order they are written in is not the reading order.
7. files.take_first: or files.take_last: keeps that many files from the sorted list (setting both is E218). With the default ascending name sort, take_first: 5 keeps the five earliest names.
8. files.on_no_match: – when nothing is left: error (default, E216), warn (log a warning and produce no rows), or skip (produce no rows quietly).

files.recursive: controls whether regex: searches subfolders (true by default). It has no effect on glob:, which searches subfolders only where the pattern has **.

- type: source
  name: orders
  config:
    name: orders
    type: csv
    glob: ./data/orders_*.csv
    exclude: ["*_partial.csv"]
    modified_after: 30d
    files:
      sort_by: name
      sort_order: desc
      take_first: 3        # the three latest names
      on_no_match: warn
    schema:
      - { name: order_id, type: string }

With glob:, regex: or paths:, each file is read as its own document: a declared sort_order is checked per file, and an Aggregate rolls up per file (see Sort order and One document per file).

Format types

The type: field inside config: selects the on-disk format. Each format has its own reference page covering its options and decoding model:

The same schema: rules apply regardless of format: the reader maps each decoded record onto the declared schema, and undeclared input fields fall under the on_unmapped policy below.

Declared-type failures

Source types are enforced before a record reaches buffering, sorting, or any downstream node. A value that cannot satisfy its authored type rejects the whole row exactly once; Clinker never substitutes the raw string, a sentinel, or an error-derived null.

Empty input has three distinct outcomes:

• an empty value declared as string remains the empty string;
• an empty non-string value declared nullable(T) becomes null;
• an empty non-string value declared as non-nullable is a type error.

Parsing does not trim whitespace, guess locale conventions, or recognize case-insensitive null sentinels. Integer overflow, invalid/out-of-range dates, decimal precision overflow, and decimal values that would need rounding to the declared scale are type errors. Accepted decimal and date values retain their declared precision.

The E126 diagnostic identifies source, file, one-based row and column, field, and declared type. Its value preview is sanitized to one line and limited to 256 rendered UTF-8 bytes; controls, bidi characters, diagnostic delimiters, backslashes, and invalid UTF-8 are explicit indivisible escape tokens. When truncated, the preview ends in one … without splitting a token or Unicode scalar and reports the original byte length. The complete original record/value is retained only in the configured DLQ. See Error Handling & DLQ for strategy and threshold behavior.

Already-decoded values obey the same declarations without hidden coercions: string admits only a string, null admits only null, and any admits every supported native value while preserving its represented value. Numeric conversions must be exact; non-finite floating-point values and decimal values that would require rounding are rejected. With multiple: true, the scalar declaration is applied to every array element. If any element fails, the whole source record is rejected and the complete original array remains available through the DLQ.

Error strategies and complete populations

fail_fast aborts on the first declared source failure. continue and best_effort require a DLQ and use the configured threshold over the complete population:

rejected records / attempted records

The run aborts only when that ratio is strictly greater than the threshold. Equality is accepted, so a threshold of 0.1 admits exactly one rejected record in a population of ten but not two. A zero threshold aborts on the first rejection; a threshold of 1.0 admits an all-rejected population for the continuing strategies.

For ordered file sources, Clinker establishes the complete attempted and rejected population before any accepted record, punctuation, downstream side effect, or output byte is released. That population is applied exactly once whether execution remains resident, spills, or fuses a downstream node. A threshold violation therefore cannot leave a committed prefix of output.

on_unmapped — undeclared input fields

The per-source on_unmapped policy decides what to do with input fields the source’s schema: block does not name. Three modes — auto_widen (default), drop, reject:

- type: source
  name: orders
  config:
    name: orders
    type: csv
    path: "./data/orders.csv"
    on_unmapped:
      mode: auto_widen     # default; other values: drop, reject
    schema:
      - { name: order_id, type: string }
      - { name: amount, type: float }

See Auto-Widen & Schema Drift for the full specification: how undeclared columns flow through each downstream node type, the include_unmapped Sink flag, E315 merge-policy mismatch, and fixed-width behavior.

Sort order

If each physical input file is pre-sorted, declare the record order so the planner can admit order-dependent strategies such as streaming aggregation:

- type: source
  name: sorted_transactions
  config:
    name: sorted_transactions
    type: csv
    path: "./data/transactions_sorted.csv"
    schema:
      - { name: account_id, type: string }
      - { name: txn_date, type: date }
      - { name: amount, type: float }
    sort_order:
      - { field: "account_id", order: asc }
      - { field: "txn_date", order: asc }

Clinker binds these fields to the declared source schema, compares the typed values by the same rule every sort uses (see How values are ordered), and verifies each physical file independently before any record from that file reaches an order-dependent consumer. The declaration never means that a multi-file source is globally sorted: the last key in one file is not compared with the first key in the next file.

on_unsorted controls the result of the first adjacent inversion:

    sort_order:
      - { field: "account_id", order: asc, null_order: last }
      - { field: "txn_date", order: desc, null_order: first }
    on_unsorted: warn

Source ordering accepts null_order: first or last; drop is rejected when the pipeline is planned because verifying order must not discard source records. The error gives one fix: delete null_order: drop and add a Transform after the Source whose whole config is the line it prints, config: { cxl: "filter not <field>.is_null()" }. With the line deleted, the Source declares its null keys last; if a file’s null keys arrive first, write null_order: first instead of deleting the line. That filter needs a field CXL can name as it is: one identifier of ASCII letters, digits and _, not starting with a digit and not a CXL keyword. For any other key, such as order id, filter or a flattened Address.City, the error prints no CXL and prints a source_name line for the column instead, such as source_name: "order id". Set the column’s name to a new identifier, add that line, use the new name wherever the pipeline names the column, and plan again: the error then prints the filter on the new name. Equal authored keys retain arrival order within the selected execution path. Clinker does not add a source identity, physical filename, or canonical-row tie-breaker.

Verification stages the complete sortable file event sequence behind the run’s memory arbitrator. It uses the existing stable resident/spill sort and bounded-fan-in merge machinery when repair spills, while preserving row identity, source/file provenance, and document context. Only flat sources and sources with one sortable frame per physical file are admitted; a format whose nested or repeated framing cannot be reordered losslessly fails during planning with a correction to remove sort_order or normalize the input.

If a downstream consumer needs one global order across all files, declare sort_order on the terminal Sink. Use enough output fields to define a total business order when byte-identical output matters.

The shorthand form is also accepted – a bare string defaults to ascending:

    sort_order:
      - "account_id"
      - { field: "txn_date", order: desc }

Watermarks

An event-time watermark declares which column on the source carries each record’s event time — the wall-clock instant the event happened, distinct from when Clinker read the row. When set, Clinker takes the column on every record, subtracts the source’s delay, and uses the result to track event-time progress so downstream time windows know when to close. The delay-corrected value is also stamped on every record as $source.event_time, the column a downstream time-windowed aggregate uses to assign records to windows.

- type: source
  name: clicks
  config:
    name: clicks
    type: csv
    path: "./data/clicks.csv"
    options:
      has_header: true
    watermark:
      column: event_ts       # must be date_time or date
      delay: 5s              # bounded out-of-order tolerance
      idle_timeout: 30s      # flip partitions to idle if quiet
    schema:
      - { name: user_id, type: string }
      - { name: event_ts, type: date_time }
      - { name: amount, type: int }

Fields:

• column (required) — the schema column whose value is each record’s event time. The column’s declared type must be date_time or date. A column: that names a field absent from schema: raises E154; a column: whose declared type is neither raises E155.

• delay (optional duration, default unset) — bounded out-of-order tolerance. Each record’s event time is shifted earlier by delay before being folded into the watermark, so the source’s effective watermark trails its observed max event time by this amount. Mirrors Flink’s BoundedOutOfOrdernessWatermarks. Without delay, the watermark advances strictly to the observed max — a single late record routes to the DLQ.

• idle_timeout (optional duration, default unset) — if a source stays quiet longer than this, it stops holding back downstream window-close progress, so windows keep closing when one source pauses. Unset means the source never goes idle.

Durations use the suffixes ms, s, m, h, d. ms is matched before the single-character s, so 500ms reads as 500 milliseconds, not 500 seconds with a stray m.

A pipeline whose aggregate declares time_window: must have a watermark.column on every upstream-reachable source. Without it, event-time progress can never advance and the window can never close — the planner rejects this with E156.

Multi-value fields

A field that holds more than one value is declared on the schema column, not on the pipeline:

schema:
  - { name: order_id, type: string }
  - { name: tags, type: string, multiple: true }

multiple: true says the column holds zero or more values of its declared type. Reading collects every occurrence of the field into one array — a single occurrence is still an array, so downstream code never has to branch on how many values happened to arrive. A field absent from a record has no column at all and resolves to null, exactly as any other absent column does. CXL sees the column as an array; the declared type: describes each element and drives coercion. The declaration describes the shape of the data, so it serves both directions: a writer that can encode repetition reads the same declaration.

The split_to_rows, split_values, and join_values blocks below accept a compact shorthand (a bare field name, or a mapping that omits defaults). To see the fully-materialized form the engine actually runs — every default spelled out — print the canonical config with clinker config --resolved. It rewrites only those shorthand blocks and leaves the rest of the file untouched.

Both ends of the declaration are checked at compile, so a shape the formats cannot carry fails before a run starts rather than mid-stream:

A multiple: true column reaching an output that cannot encode it is E359; one on a source that cannot produce it is E361. E359 covers an output’s own schema: block too — the attribute is direction-neutral, but the remaining writers do not encode repetition yet, so declaring it on such a sink would be accepted and ignored. Run clinker explain --code E361 for the full remediation of either.

csv and fixed_width read a multiple: true column through a split_values entry. Neither wire format repeats a field, but a cell’s text may hold several values separated by a delimiter. Declare that delimiter with a split_values entry and the reader parses the cell into the array the column holds. A multiple: true column no entry covers is rejected by E361 — the reader would have no delimiter and deliver the raw cell; either add the entry or leave the column single-valued and split it in a transform (tags.split(";")). The entry is read only on a single-schema source: a multi-record source of either format runs a backend that does not consume it. On the output side, a CSV sink joins a multiple: field into one delimited cell with join_values (defaulting to ; / on_conflict: error); fixed_width output is still pending (#918).

A split_values entry also recovers a CSV cell a sink wrote under join_values on_conflict: escape or encode_json: add escape: "\\" to un-escape an escaped delimiter, or json: true to read the whole cell as an embedded JSON array.

The segment formats are a permanent no, not a pending one. Repetition there is a positional coordinate rather than a list: a repeated composite is written as two axes interleaved in one element (11:B:1^12:B:2), which a flat array cannot represent without losing the component axis. The faithful shape is one column per coordinate — for HL7, that is what options.split_fields produces, with a writer that reassembles the wire field byte-for-byte.

One record per value: split_to_rows

split_to_rows fans a record out to one record per occurrence of a repeated field. Each entry is either a bare field name or a full mapping, and the two forms mix freely in one list:

- type: source
  name: invoices
  config:
    name: invoices
    type: json
    path: "./data/invoices.json"
    schema:
      - { name: invoice_id, type: int }
      - { name: customer, type: string }
      - { name: line_item, type: string }
      - { name: line_amount, type: float }
      - { name: line_no, type: int }
    split_to_rows:
      - tags                      # shorthand: field name, all defaults
      - field: line_items         # full form
        keep_empty: true
        mode: extract
        position_column: line_no
    max_output_rows_per_input: 10000

The field is named as it appears in the input document, not as the schema exposes it: a column declared source_name: is addressed by that source_name. The same rule applies to split_values below.

keep_empty defaults to true. A record whose field holds an empty array, or carries no such field at all, is emitted with that field unset rather than disappearing. Several widely used engines drop the record instead; a vanished row is the costliest failure mode there is, so dropping is opt-in here.

mode: extract (the default) makes the occurrence the record: its own fields are lifted out from under the field name and every field outside the group is merged onto each output. {"orders": [{"id": 1}]} yields a top-level id, and repeated <Item><name> children yield name. When lifting lands an occurrence’s field on a name an outside field already occupies, the occurrence wins — it is the record, so its own value is not shadowed by the parent it was merged with. A position_column wins over both: you named it, so a field of the same name inside or outside the occurrence gives way to the index.

mode: split preserves the record shape: the occurrence’s fields keep their dotted path (orders.id, Item.name) and each output carries exactly one occurrence.

Entries apply in declaration order, so two entries multiply. Declaring the same field twice is rejected at compile (E358), as is fanning out a field the schema also declares multiple: true — the attribute collects the occurrences into one array, the fan-out spends them one per record, and a field cannot be both. On an XML source, two entries may not name nested element groups either (Item and Item.part): that reader assigns each element to one occurrence group by document position, and a nested pair leaves the inner group’s membership ambiguous.

max_output_rows_per_input is valid only alongside a non-empty split_to_rows block and bounds that cumulative product for one original JSON object or XML record element. Omit it, or set it to 0, for no ceiling. With a positive value N, the reader emits exactly the first N records in the same declaration order shown above. If an N+1 row is attempted, the reader stops that input and reports an expansion_limit_exceeded source failure. Under error_handling.strategy: continue, the complete decoded original input representation is routed to the DLQ; the first N records remain valid output. This is not silent truncation: the run records one explicit rejection naming the field, the configured ceiling, and the exact first violating count (N+1). Under fail_fast, the run fails at that boundary.

The check is lazy: the reader never materializes or pre-counts the Cartesian product. Its cursor holds the original input, the declared occurrence lists, and one output record. This source setting is separate from a Transform’s max_expansion; when both surfaces fan out, each enforces its own per-input boundary.

A JSON source accepts a nested pair to produce a two-level expansion, but only when the outer entry declares mode: split. mode: extract lifts the occurrence’s own keys to the top level, which removes the dotted path the inner entry addresses — the inner entry would then match nothing and fan nothing out, so the pairing is rejected (E358):

    split_to_rows:
      - { field: orders, mode: split }
      - { field: orders.items, mode: split }

Several values in one cell: split_values

split_values parses a delimited cell into the several values a multiple: column holds. It takes the same bare-name-or-mapping shorthand:

    split_values:
      - tags                      # shorthand: default delimiter `;`
      - field: codes              # full form
        delimiter: "|"
    schema:
      - { name: tags,  type: string, multiple: true }
      - { name: codes, type: string, multiple: true }

The delimiter defaults to ;. A split_values field the schema does not declare multiple: true is rejected at compile (E358): splitting produces several values, and only a multi-value column can hold them. So is an entry naming a column’s exposed name when that column reads a differently-named input field — the split runs against the document’s own field names, so name the source_name.

The entry is read by the JSON and XML readers, and — on a single-schema source — by the CSV and fixed-width readers. On a multi-record CSV or fixed-width source, or on any segment format, declaring it is rejected (E358) rather than silently ignored: those readers are never handed it, so the cell would arrive unsplit with nothing to say so.

Migrating from array_paths

array_paths: was the earlier form of these declarations. It is no longer read, and a source still carrying it is rejected at compile (E360) rather than running with the fan-out silently dropped. An explode path becomes a split_to_rows: entry, a delimited cell becomes a split_values: entry, and a path kept as an array becomes multiple: true on the schema column.

mode: extract (the default) reproduces the old projection — the element’s own fields lifted to the top level of each output record. It does not reproduce the old cardinality: explode dropped a record whose array was empty, while keep_empty defaults to true here and keeps it with the element’s fields unset. Add keep_empty: false to the entry to migrate row-for-row.

Format notes

split_to_rows is honored by the JSON and XML readers, over a file path and over a rest response body alike. split_values is honored by those two and also by the single-schema CSV and fixed-width readers. Declaring a knob on a reader that is never handed it — split_to_rows on any delimited-cell or segment format, split_values on a multi-record or segment source — is rejected at compile (E358) rather than accepted and inert, and a multiple: true column no split_values entry can cover is rejected by E361.

Composition body files are gated by the same four checks as the pipeline that calls them.

JSON — the field names the key holding the array (line_items, or order.line_items for an array nested under an object). A field present but holding a single object or scalar rather than an array counts as one occurrence, and is projected exactly as a one-element array would be — many producers unwrap a lone element, and XML cannot express the difference at all, so the two readers agree on this. A field that is absent, holds an empty array, or is explicitly null has no occurrence and is governed by keep_empty: an explicit null is how many producers write “no value”, and it counts as none rather than as one. For the same reason a multiple: true column holding an explicit null stays null rather than becoming [null], so size() over it reads the same as it does for a field the document omits.

XML — the field is the repeated child element’s dotted path relative to the record element (Item, or Items.Item when nested). Repetition and absence are indistinguishable in XML, so a record with no occurrence of the element is governed by keep_empty exactly as an empty array is. See XML Format for the full rules.

Transform Nodes

Transform nodes apply CXL expressions to each record, producing new fields, filtering records, or both. They process one record at a time in streaming fashion with constant memory overhead.

Basic structure

- type: transform
  name: enrich
  input: customers
  config:
    cxl: |
      emit full_name = first_name.concat(" ", last_name)
      emit tier = if lifetime_value >= 10000 then "gold" else "standard"
      filter status == "active"

The cxl: field is required and contains a CXL program. The three core CXL statements for transforms are:

• emit – adds a field to the record, or replaces the field of the same name. The record’s other input fields are carried through to downstream nodes without being emitted; a Sink with include_unmapped: false is the one place that narrows them (see Sink Nodes).
• filter – drops records that do not match the boolean condition.
• let – binds a local variable for use in subsequent expressions (not emitted).

    cxl: |
      let margin = revenue - cost
      emit product_id = product_id
      emit margin = margin
      emit margin_pct = if revenue > 0 then margin / revenue * 100 else 0
      filter margin > 0

Analytic window

The analytic_window field enables cross-source lookups by joining a secondary dataset into the transform. The secondary source is loaded into memory and indexed by the join key.

- type: transform
  name: enrich_orders
  input: orders
  config:
    analytic_window:
      source: products
      on: product_id
      group_by: [product_id]
    cxl: |
      emit order_id = order_id
      emit product_name = $window.first().product_name
      emit quantity = quantity
      emit line_total = quantity * price

The $window.* namespace provides access to the windowed data. Functions like $window.first().<field>, $window.last().<field>, and $window.count() operate over the matched group. first() and last() return a whole record, so name the field to read; without one they return null. See Window Functions.

Validations

Declarative validation checks can be attached to a transform. They run against each record and either route failures to the DLQ (severity error) or log a warning and continue (severity warn).

- type: transform
  name: validate_orders
  input: raw_orders
  config:
    cxl: |
      emit order_id = order_id
      emit amount = amount
      emit email = email
    validations:
      - field: email
        check: "not_empty"
        severity: error
        message: "Email is required"
      - check: "amount > 0"
        severity: warn
        message: "Non-positive amount"
      - field: order_id
        check: "not_empty"
        severity: error

Validation fields

Expansion cap (max_expansion)

When a transform body contains an emit each statement, every input record can fan out into multiple output records. The max_expansion field caps how many output records a single input record may produce – a safety bound against unexpectedly large arrays.

- type: transform
  name: explode_items
  input: orders
  config:
    max_expansion: 5000      # default: 10000
    cxl: |
      emit each it in items {
        emit order_id = order_id
        emit sku = it["sku"]
        emit price = it["price"]
      }

If a single input record’s emit each block produces more than max_expansion output records, the originating record routes to the DLQ with category expansion_limit_exceeded instead of producing a truncated or unbounded result. No partial output is emitted for that record – the cap is enforced eagerly so the writer never sees records from a runaway expansion.

When to tune

• Lower (e.g. 100, 1000) when input arrays are bounded by a known business rule and you want hostile or malformed input to surface as a DLQ entry rather than as a flood of downstream records.
• Higher (e.g. 100000, 1000000) when legitimate input carries large arrays – for example, an order with a long line-item list or an event carrying a per-second pricing curve.

The DLQ category expansion_limit_exceeded is distinct from generic CXL evaluation failures, so DLQ-side filters and metrics can target expansion runaway specifically. See Error Handling & DLQ for the wider DLQ contract.

Batch size (batch_size)

A streaming-eligible transform hands its output downstream in bounded batches rather than accumulating the whole stage before the next stage runs. batch_size sets how many events (records plus document-boundary punctuations) a batch holds. A per-transform batch_size overrides the pipeline-level pipeline.batch_size for this one stage; omit it to inherit the pipeline value (or the built-in default of 2048).

- type: transform
  name: enrich
  input: orders
  config:
    batch_size: 512         # override pipeline.batch_size for this stage
    cxl: |
      emit order_id = order_id
      emit total = quantity * unit_price

A batch_size of 0 is rejected at config load (a zero-event batch never flushes). Smaller batches lower the in-flight memory of a streaming stage at the cost of more per-batch bookkeeping; larger batches amortize the bookkeeping at the cost of a larger live working set. The default suits typical record widths — tune it only when a profiling run shows a streaming stage’s per-batch footprint matters. See Streaming vs. Blocking Stages for which stages stream and which fully materialize.

Log directives

Log directives declare bounded structured diagnostic events during transform execution:

- type: transform
  name: process
  input: validated
  config:
    cxl: |
      emit id = id
      emit result = compute(value)
    log:
      - name: transform.record_processed
        level: info
        when: per_record
        every: 1000
        message: "Processed record"
        fields: [id]
      - name: transform.record_failed
        level: warn
        when: on_error
        message: "Record failed processing"
      - name: transform.started
        level: debug
        when: before_transform
        message: "Starting transform"

Log directive fields

A transform may declare at most 32 events and request at most 256 fields in aggregate across them. Event names and field selectors use the same grammar as deployment field policy: dot-separated segments beginning with an ASCII letter or underscore, followed by ASCII letters, digits, or underscores.

fields is the only channel by which record data reaches an event — message is static text. A selector naming a field the incoming record does not carry contributes nothing, so a directive whose selectors all miss would publish an event with no attributes at all, which reads exactly like a run whose records were empty.

Clinker refuses that when the pipeline compiles (E374). The rejection names the selector, lists the columns the input row does carry, and — when your spelling is close to one of them — gives you the corrected fields: line to paste:

[E374] transform `enrich` log[0].fields requests `orderId`, which the input
record does not carry; the upstream row has `order_id`, `amount`, `region` —
write `fields: [order_id]`

Selectors bind against the transform’s input row, for per_record and on_error alike: dispatch fires before this transform’s own cxl: block, so a column the transform produces cannot be requested. Request the columns it reads instead.

The check decides what the declared schema can decide. A column that reaches the transform through an open composition port is not visible to it, and a selector naming one of those is still checked only at run time — counted in the run’s admission accounting under the missing-field total.

One event name, one set of fields

An event name is the identity a collector groups records by, and it carries no node identity of its own. Two transforms may emit the same event — that is how one thing that happens in several places is reported as one thing — but they have to declare the same fields, or the collector receives two record shapes under one name and nothing downstream can separate them.

Clinker checks this across the whole pipeline, composition bodies included, and refuses a disagreement with E375:

[E375] transform "shape" in composition "customer_enrichment": `log` event
"transform.customer_seen" is also declared by transform "seen" with different
fields; one event name carries one set of fields — give both declarations the
same `fields`, or give one of them its own event name

A composition used twice is not a conflict: both instances declare the same event with the same fields, and agreement is what the rule asks for.

Logging only the records you care about

every thins a per-record event by count. condition selects it by content — use it when the interesting records are rare and you want all of them rather than every thousandth record:

    log:
      - name: transform.large_order
        level: info
        when: per_record
        every: 1
        condition: "amount > 1000"
        message: "large order"
        fields: [order_id, amount]

The two compose: every is applied first, then condition, so every: 100 with a condition logs every hundredth record that also matches.

A condition is CXL, checked when the pipeline compiles. It must resolve to a boolean, and it is evaluated against the transform’s input record — the one that arrived, before this transform’s own cxl: block runs. A field the transform only produces is therefore not in scope; write the condition in terms of the fields the transform reads.

A condition decides only whether an event fires. It cannot add anything to one: the values that leave the process are still exactly the fields you requested, each still subject to deployment policy. Narrowing a condition can never widen what is exported.

Transform declarations name events, request fields, and may gate a per-record event on its own input; they do not choose a destination, credentials, routing, redaction, or sampling policy. Each requested event-field pair is denied unless deployment observability policy explicitly allows, hashes, or replaces it. Telemetry delivery is bounded and best effort and cannot change transform results or published output — including a condition that fails to evaluate, which drops its event rather than failing the run.

Every transform event also carries the fixed correlation fields execution_id, batch_id, and pipeline_name. Unlike requested record fields, these are not default-deny and are not gated by field_policy: they are engine-supplied identity that never derives from a source, and they are what makes an exported event joinable to the machine stream and to the lineage events. A deployment that allows an event without also writing three correlation rules still gets telemetry it can correlate.

Because they are exported verbatim, choose a --batch-id that is safe to send to your collector — an identifier, not a tenant name or anything else you would not want retained there. A field_policy rule naming one of these three fields does not redact it. Source paths, records, secrets, and raw error text are never implicit attributes.

The former log_rule directive key and pipeline-level log_rules block are rejected. Move event identity and safe field requests into the transform’s log: entries as shown above; keep routing, privacy, credentials, and sampling in deployment policy.

Complete example

- type: source
  name: employees
  config:
    name: employees
    type: csv
    path: "./data/employees.csv"
    schema:
      - { name: employee_id, type: string }
      - { name: first_name, type: string }
      - { name: last_name, type: string }
      - { name: department, type: string }
      - { name: salary, type: int }
      - { name: hire_date, type: date }

- type: transform
  name: enrich_employees
  description: "Compute display name and tenure"
  input: employees
  config:
    cxl: |
      emit employee_id = employee_id
      emit display_name = last_name.concat(", ", first_name)
      emit department = department.upper()
      emit salary = salary
      emit annual_bonus = if salary >= 80000 then salary * 0.15
        else salary * 0.10
    validations:
      - field: employee_id
        check: "not_empty"
        severity: error
        message: "Employee ID is required"
      - check: "salary > 0"
        severity: warn
        message: "Salary should be positive"
    log:
      - name: transform.employee_processed
        level: info
        when: per_record
        every: 5000
        message: "Processing employees"
        fields: [employee_id]

Route Nodes

Route nodes split a stream of records into named branches based on CXL boolean conditions. Each branch becomes an independent output port that downstream nodes can wire to using port syntax.

Interactive companion: the Route and Merge explainer shows, record by record, which conditions are checked and where each record goes, and how a Merge rejoins the branches.

Basic structure

- type: route
  name: split_by_value
  input: orders
  config:
    mode: exclusive
    conditions:
      high: "amount.to_int() > 1000"
      medium: "amount.to_int() > 100"
    default: low

This creates three output ports: split_by_value.high, split_by_value.medium, and split_by_value.low.

Conditions

The conditions: field is an ordered map of branch names to CXL boolean expressions. Each expression is evaluated against the incoming record.

    conditions:
      priority: "urgency == \"high\" and amount > 500"
      standard: "urgency == \"medium\""
      bulk: "quantity > 100"
    default: other

Condition keys become the port names used in downstream input: wiring.

Compile-time checking

Branch conditions are typechecked when the pipeline is compiled – the plan-building pass you trigger with clinker run pipeline.yaml --explain or a bare --dry-run (see Validation & Dry Run). Each condition is checked against the concrete column types of the route’s input, so a condition that references an unknown column or compares incompatible types fails at compile time rather than partway through a run.

Compile failures surface as a CXL diagnostic keyed to the failure class – E202 for a branch condition that does not parse, E203 for one that references an unknown column, and E200 for one that compares incompatible types – and name the offending branch, for example split_by_value (branch high), so in a multi-branch route the error points at the specific branch rather than just the route node.

Default branch

The default: field is required. Records for which no condition is true are routed to the default branch. A condition whose result is null counts as not true.

A condition that fails to evaluate is not “no match”. Under error_handling.strategy: continue the record is dead-lettered and takes no branch, not even one whose condition held, and not the default; under fail_fast the run stops. In exclusive mode the conditions after the first true one are never evaluated, so a condition further down cannot fail for that record. See An evaluation error is never false.

Routing modes

Exclusive (default)

In exclusive mode, conditions are evaluated in declaration order and the first matching condition wins. A record appears in exactly one branch. Order matters – put more specific conditions first.

    mode: exclusive
    conditions:
      vip: "lifetime_value > 100000"
      high: "lifetime_value > 10000"
      medium: "lifetime_value > 1000"
    default: standard

A customer with lifetime_value = 50000 is not over 100000, so vip is not true; high is the first true condition and wins. A customer with lifetime_value = 150000 is true for all three conditions, but goes to vip alone, because vip is checked first. Listing medium first would send both customers to medium.

Inclusive

In inclusive mode, all matching conditions route the record. A single record can appear in multiple branches simultaneously.

    mode: inclusive
    conditions:
      needs_review: "amount > 10000"
      flagged: "status == \"flagged\""
      international: "country != \"US\""
    default: standard

A flagged international order over 10000 would appear in needs_review, flagged, and international – three copies routed to three branches.

Downstream wiring

Downstream nodes reference route branches using port syntax: route_name.branch_name. The default branch is reached the same way, by its name. Several nodes can read the same branch; each receives every record on it.

A node whose own name equals a branch name can also reference the Route by its bare name and receives that branch: a Sink named high with input: classify reads classify.high.

- type: route
  name: classify
  input: transactions
  config:
    mode: exclusive
    conditions:
      high: "amount > 1000"
      medium: "amount > 100"
    default: low

- type: transform
  name: high_value_processing
  input: classify.high
  config:
    cxl: |
      emit txn_id = txn_id
      emit amount = amount
      emit review_flag = true

- type: transform
  name: standard_processing
  input: classify.medium
  config:
    cxl: |
      emit txn_id = txn_id
      emit amount = amount

- type: sink
  name: low_value_out
  input: classify.low
  config:
    name: low_value_out
    type: csv
    path: "./output/low_value.csv"

Constraints

• Give every condition a distinct name; each name is a separate port.
• Give default a name that no condition uses. This is not currently checked when the pipeline is planned: a default with the same name as a condition is folded into that condition’s branch, so its records cannot be told apart from the matches.
• Declare at least one condition. A Route with none is not rejected today; every record takes the default branch.

Complete example

pipeline:
  name: order_routing

nodes:
  - type: source
    name: orders
    config:
      name: orders
      type: csv
      path: "./data/orders.csv"
      schema:
        - { name: order_id, type: int }
        - { name: region, type: string }
        - { name: amount, type: float }
        - { name: priority, type: string }

  - type: route
    name: by_region
    input: orders
    config:
      mode: exclusive
      conditions:
        domestic: "region == \"US\" or region == \"CA\""
        emea: "region == \"UK\" or region == \"DE\" or region == \"FR\""
        apac: "region == \"JP\" or region == \"AU\" or region == \"SG\""
      default: other

  - type: sink
    name: domestic_orders
    input: by_region.domestic
    config:
      name: domestic_orders
      type: csv
      path: "./output/domestic.csv"

  - type: sink
    name: emea_orders
    input: by_region.emea
    config:
      name: emea_orders
      type: csv
      path: "./output/emea.csv"

  - type: sink
    name: apac_orders
    input: by_region.apac
    config:
      name: apac_orders
      type: csv
      path: "./output/apac.csv"

  - type: sink
    name: other_orders
    input: by_region.other
    config:
      name: other_orders
      type: csv
      path: "./output/other_regions.csv"

Merge Nodes

Merge nodes concatenate multiple upstream branches into a single stream. They are the counterpart to route nodes – where a route splits one stream into many, a merge joins many streams back into one.

Merge is for streamwise concatenation of inputs that share a schema. For record-level joining across inputs that have different schemas, see Combine Nodes.

Interactive companion: the Route and Merge explainer shows how each mode mixes its inputs, and what an inclusive Route’s copies look like after a Merge.

Basic structure

- type: merge
  name: combined
  inputs:
    - east_data
    - west_data
  config: {}

Note the key differences from other node types:

• Uses inputs: (plural), not input: (singular).
• The config: block is empty – all wiring is on the node header.
• Using input: (singular) on a merge node is a parse error.

Wiring

The inputs: field is a list of upstream node references. These can be bare node names or port references from route nodes:

- type: merge
  name: rejoin
  inputs:
    - process_high
    - process_medium
    - classify.low           # Port syntax for a route branch
  config: {}

Downstream nodes wire to the merge as a normal single-input reference:

- type: sink
  name: final_output
  input: rejoin
  config:
    name: final_output
    type: csv
    path: "./output/combined.csv"

Modes

Merge’s cross-input ordering discipline is selected by config.mode. Two modes exist; concat is the default.

concat (default)

Predecessor records drain in declaration order: inputs[0] flows to output first, then inputs[1], then inputs[2], and so on. Within a single predecessor, its arrival order is preserved. Output is reproducible run-to-run for the same predecessor paths.

- type: merge
  name: combined
  inputs: [east, west]
  config:
    mode: concat

interleave

Records flow to output as they become available from any predecessor. Each input’s arrival order is preserved; cross-input order follows wall-clock arrival and is not promised.

- type: merge
  name: combined
  inputs: [east, west]
  config:
    mode: interleave

Seeded interleave — interleave_seed:

Snapshot tests and benchmarks that need reproducible cross-input ordering can opt into a deterministic schedule:

- type: merge
  name: combined
  inputs: [east, west]
  config:
    mode: interleave
    interleave_seed: 42

With a seed, the cross-input order is reproducible from run to run regardless of upstream timing. To get there, the Merge reads all of its inputs into memory before emitting, so a seeded interleave buffers more than the other modes — use it for tests and benchmarks, not high-volume production merges.

Choosing a mode

For high-volume merges, prefer concat or unseeded interleave — both stream their inputs and let a slow downstream consumer naturally throttle the upstream readers, so memory stays bounded. A seeded interleave does not, because it buffers everything first.

Record ordering

Records arrive in the order described by the mode in use — see Modes and Choosing a mode above. Merge does not promote matching per-input sort_order declarations into one global sort: concatenating two independently sorted files can still put a high key before a lower key at the input boundary.

For concat and seeded interleave, exact sequence is a supported oracle for the same input paths and configuration. For unseeded interleave, compare the decoded record multiset and aggregate values; snapshotting incidental cross-input arrival order would assert behavior Clinker does not promise. If you need one sorted sequence regardless of Merge mode, declare sort_order on the downstream Sink. That terminal sort uses only the authored keys and does not invent a hidden identity tie-breaker.

Use cases

Reuniting route branches

The most common pattern is routing records through different processing paths and then merging them back together:

- type: route
  name: classify
  input: orders
  config:
    mode: exclusive
    conditions:
      high: "amount > 1000"
    default: standard

- type: transform
  name: process_high
  input: classify.high
  config:
    cxl: |
      emit order_id = order_id
      emit amount = amount
      emit surcharge = amount * 0.02
      emit tier = "premium"

- type: transform
  name: process_standard
  input: classify.standard
  config:
    cxl: |
      emit order_id = order_id
      emit amount = amount
      emit surcharge = 0
      emit tier = "standard"

- type: merge
  name: all_orders
  inputs:
    - process_high
    - process_standard
  config: {}

- type: sink
  name: result
  input: all_orders
  config:
    name: result
    type: csv
    path: "./output/all_orders.csv"

A Merge passes on every record from every input and does not remove duplicates. Branches of an inclusive Route can carry the same record, and rejoining them puts that record in the output once per branch it took. Use an exclusive Route when each record should come out once.

Unioning multiple sources

Merge nodes can combine records from multiple source files that share the same schema:

- type: source
  name: jan_sales
  config:
    name: jan_sales
    type: csv
    path: "./data/sales_jan.csv"
    schema:
      - { name: sale_id, type: int }
      - { name: amount, type: float }
      - { name: region, type: string }

- type: source
  name: feb_sales
  config:
    name: feb_sales
    type: csv
    path: "./data/sales_feb.csv"
    schema:
      - { name: sale_id, type: int }
      - { name: amount, type: float }
      - { name: region, type: string }

- type: merge
  name: all_sales
  inputs:
    - jan_sales
    - feb_sales
  config: {}

- type: aggregate
  name: totals
  input: all_sales
  config:
    group_by: [region]
    cxl: |
      emit total = sum(amount)
      emit count = count(*)

Schema constraints across inputs

Merge concatenates streams positionally against the merge node’s output_schema (taken from the first input). Every input must therefore agree on column shape — same column names, same on_unmapped policy, same correlation_key set.

Disagreement on the $widened auto_widen sidecar (one source uses auto_widen, another uses drop / reject) fails compile with E315. See Auto-Widen & Schema Drift → E315 for the full diagnostic shape and remediation.

Combine Nodes

Combine nodes are the N-ary record-combining operator. Every input is declared up front and bound to a qualifier; the where: expression matches records across inputs using qualified field references (e.g. orders.product_id == products.product_id); the cxl: body shapes the output row.

Combine is distinct from merge: merge concatenates upstream branches that share a schema, while combine joins records across inputs that have different schemas.

Interactive companion: the Combine playground shows, driver row by driver row, which build rows where: matches and what match: and on_miss: do with them, including range joins.

Basic structure

- type: combine
  name: enrich
  input:
    orders: orders         # qualifier: upstream node name
    products: products
  config:
    where: "orders.product_id == products.product_id"
    match: first
    on_miss: null_fields
    cxl: |
      emit order_id = orders.order_id
      emit product_name = products.product_name
      emit amount = orders.amount
    propagate_ck: driver

Note the differences from other node types:

• Uses input: as a map, binding qualifier names to upstream node references. Other nodes use input: as a single string or inputs: as a list of strings.
• Every field reference inside where: and cxl: must be qualified (<qualifier>.<field>). Bare field names are a compile error.
• Using inputs: (plural list) on a combine node is a parse error.

Wiring

Each entry in the input: map binds a qualifier to an upstream node:

  input:
    orders: orders                  # qualifier "orders" -> source node "orders"
    products: products
    high_priority: classify.high    # qualifier "high_priority" -> route port

Qualifiers are local names used inside where: and cxl:; they do not need to match the upstream node name. Upstream references can be bare node names or port references from a route node.

Iteration order in the input: map is preserved and used as the default driver-selection order (see Choosing the driving input below).

Configuration fields

The where: predicate

The where: expression is a CXL boolean expression evaluated for every candidate record pair across inputs. It must contain at least one cross-input equality – an equality with field references from two different inputs:

  where: "orders.product_id == products.product_id"

Compound predicates combine multiple conjuncts with and. Each conjunct is classified by the planner:

• Equi conjunct – a cross-input equality (a.x == b.y). Drives the hash lookup or sort-merge join.
• Range conjunct – a cross-input ordered comparison (a.start <= b.ts and b.ts <= a.end). Handled by a range join (IEJoin), whether or not an equi conjunct also links the same two inputs.
• Residual conjunct – any other CXL predicate (intra-input filter, function call, etc.). Applied as a post-filter after the equi/range match.

  where: |
    orders.product_id == products.product_id
    and orders.amount >= 100
    and products.region == "us-east"

Above: the equi conjunct drives the join; orders.amount >= 100 and products.region == "us-east" are applied as residuals.

Every combine predicate must carry at least one cross-input equality or range conjunct. A predicate with neither — a pure residual with no decomposable cross-input comparison — is rejected at plan time with diagnostic E313; there is no supported execution strategy for it. Pure-range predicates without an equi conjunct are fully supported via IEJoin.

Non-orderable range keys

A range conjunct compares values that must be orderable at runtime: integers, finite floats, exact decimals, dates, and datetimes. When a record’s range key evaluates to a non-orderable value — SQL NULL, a non-finite float (NaN/infinity), or any other type — that record can never satisfy the range comparison, so it is routed out of the range match rather than joined:

Decimal and mixed-numeric range keys. Exact fixed-point decimal range keys are fully supported on every join strategy — a monetary band join (amount >= tier.floor and amount < tier.ceiling) matches correctly. A range conjunct that mixes an integer and a float operand across the two inputs is also supported: the integer is compared as a float, exactly as the >=/< operators compare it elsewhere.

Decimal range keys are placed on a shared fixed-point grid with up to 18 fractional digits and an integer magnitude up to roughly 1.7 × 10²⁰ — well beyond any realistic monetary value. A decimal range value outside that grid (more than 18 fractional digits, which would truncate, or a magnitude that would overflow the grid) is not silently dropped: the run stops with diagnostic E326 naming the combine, so a wrong or empty result is never emitted. Rescale or narrow the compared values if you hit it.

Datetime range keys. date and datetime range keys compare at their native resolution — a datetime to the nanosecond, across the full representable calendar (no microsecond rounding; instants before 1677 or after 2262 stay exact). Two timestamps that differ only below the microsecond therefore match, sort, and group as distinct instants on every join strategy, so a sub-microsecond as-of or band lookup neither drops a boundary match nor merges two near-simultaneous events.

abs/min/max/clamp and rejected range keys (E327). abs, min, max, and clamp return the numeric supertype (int | float). When both operands of the range conjunct recover the same concrete type — abs(int) >= abs(int), or a min/max/clamp whose result can only be one type (all-int or all-float, recovered through nested calls and arithmetic such as abs(a.x + 1) or abs(min(a.i, b.i))) — the axis is exactly as safe as a plain matching-typed key and the join runs normally. Otherwise the conjunct cannot be reduced to one exact numeric axis and is rejected at plan time with E327, rather than routed to the join where it could silently drop rows or mismatch. E327 fires when such an operand stays genuinely ambiguous — a mixed pair like abs(int) >= abs(float), or a per-row-ambiguous result like min(int, float) that may return either the integer or the float operand — or on a non-orderable pairing such as a string comparison or a decimal compared against a float/numeric. Compare a matching-typed range key instead. (A supported int/float/decimal/date/datetime range key, including a mixed int/float field pair, never triggers E327.)

• A driver record with a non-orderable range key is treated as a zero-match driver and handled by on_miss (null_fields / skip / error).
• A build record with a non-orderable range key is dropped (it can match nothing).

This is a runtime routing decision on the data, distinct from the plan-time E313 rejection above (which is about the predicate shape, not the values).

Match modes

match: first

Emit one output row per driver record, using the first matching build-side record. Standard 1:1 enrichment. Default.

“First” means the earliest in the build input’s arrival order: of the build records that match a driver, the one the build input delivered first. The same order sets the row order of match: all within a driver and the element order of a match: collect array. It holds identically for every join strategy the planner may pick and whatever shape the where: predicate has. One exception remains. When the build input exceeds the memory limit, the Combine sets part of it aside on disk and splits that part into smaller pieces until each fits. If a piece still does not fit after splitting, the Combine processes it in chunks and currently decides first, collect and on_miss once per chunk rather than once per driver, so a driver can receive more than one first row or more than one collect row. A piece fails to fit in three cases: the build rows sharing one key value do not fit on their own; the build input has already been split into the maximum of 4,096 pieces; or splitting a piece again leaves most of its rows together, as happens when many rows share a key value or their key values happen to land in the same piece. So the exception can reach a driver even when no single key value is over the limit, for example with a large build input under a tight memory limit. This is a known defect.

• To pick a different record, order the build input upstream. For example, to enrich each driver with the latest price, deliver the build input sorted descending on its date, such as with a descending sort_order on the build Source. A Combine has no ordering option of its own.
• A correlation key orders its Source. Declaring correlation_key on the build Source makes the planner sort that Source’s rows by the key before they reach the Combine, unless its declared sort_order already begins with the key. That order is what “first” follows. See Correlation keys.
• An unseeded interleave Merge has no fixed order. When the build input is such a Merge, its cross-input order follows arrival at run time, so “first” follows that unfixed order too. Use concat, or a seeded interleave, for a build input whose order must be the same on every run.

  config:
    where: "orders.product_id == products.product_id"
    match: first
    cxl: |
      emit order_id = orders.order_id
      emit product_name = products.product_name

The where: predicate selects the match; the cxl: body is a post-match projection that runs once on the chosen build record. Selection and projection are separate steps: if the body filters the row out (a filter that fails, or a body that emits nothing), that one output row is dropped. The combine does not fall back to a later matching build, and the driver is not treated as unmatched — it matched the predicate, the body just produced no row. on_miss (below) never fires for such a driver; it fires only when the predicate matched nothing at all. This holds identically for every join strategy the planner may pick.

Behavior change. This is a change in observable output for existing pipelines whose where: predicate carries a range, equi+range, or single-inequality comparison — the shapes the planner runs as a sort-merge join or an IEJoin (both the pure-range block-band path and the equi+range hash-partitioned path). On any of those three strategies, a driver that matched the predicate but whose body skipped every candidate was previously routed to on_miss — firing null_fields, skip, or error. It now silently produces no row, matching the pure-equality strategies (in-memory hash and grace hash), which already behaved this way and are unchanged. A pipeline that relied on the old routing (for example, on_miss: error tripping on a body-skipped driver) no longer sees it.

When where: fails to evaluate. A where: that raises an error for a candidate, such as a division by zero, has not said whether that candidate matches. It is neither a match nor a non-match. Under match: first the earliest candidate that is not a non-match decides. If it matched, the driver is enriched with it. If its where: failed, the driver is dead-lettered with that build row and writes no output row, even when a later candidate would match: taking the later one would publish a row that depends on an evaluation that failed. A candidate after the deciding one is not part of the driver’s result, so its failure is never written. This holds on every join strategy.

match: all

Emit one output row for every matching build-side record. 1:N fan-out – if a driver record matches three build records, three rows are emitted, in the build input’s arrival order (see match: first).

  config:
    where: "employees.department == benefits.department"
    match: all
    cxl: |
      emit employee_id = employees.employee_id
      emit benefit = benefits.benefit_name

Each output row depends only on its own pair, so a candidate whose where: fails to evaluate is dead-lettered with its build row while the driver’s other matches are still emitted.

match: collect

Gather every matching build-side record into a single Array-typed field on the output row. The driver record appears once; the build matches are aggregated into an array, in the build input’s arrival order (see match: first). The cxl: body must be empty under collect – the combine node synthesizes the output as { driver fields..., <build_qualifier>: Array }.

  config:
    where: "orders.product_id == products.product_id"
    match: collect
    cxl: ""

A per-group entry limit of 10,000 prevents unbounded growth.

A collect row states the complete set of a driver’s matches. If any candidate’s where: fails to evaluate, that set is unknown, so the driver writes no row, neither a partial array nor an empty one; each failing candidate is dead-lettered with its build row. Candidates past the 10,000-entry limit are still checked, so every failure among them is dead-lettered too.

Use collect when you need the set of matches as a single structured value; use all when you need a flat row per match.

Unmatched records (on_miss)

on_miss controls what happens to driver records with zero predicate matches — drivers for which no build-side record satisfied where:, because every candidate evaluated it to false or null, or because there was no candidate at all. Two kinds of driver are not misses and never reach on_miss, under any match mode and on every join strategy:

• A driver that matched the predicate but whose cxl: body skipped or failed the row (see match: first). It simply produces no output row for that match, and a body failure is dead-lettered.
• A driver any of whose candidates failed to evaluate where:. Each failure is dead-lettered, and whatever on_miss says, it does not fire: on_miss: error does not stop the run and on_miss: null_fields writes no null-filled row, because the driver was never shown to have no match.

  config:
    where: "orders.product_id == products.product_id"
    on_miss: skip

on_miss: error is useful for strict referential integrity where any miss should halt processing. on_miss: skip is the inner-join shape. on_miss: null_fields is the left-join shape and the default.

Composite keys

Chain multiple cross-input equalities with and:

  config:
    where: |
      sales.department == targets.department
      and sales.region == targets.region
    cxl: |
      emit department = sales.department
      emit region = sales.region
      emit actual = sales.amount
      emit goal = targets.goal

All conjuncts must hold for a record pair to match.

Multi-input combine (three or more)

Combine accepts any number of inputs. Each pair of inputs that should be related needs an explicit cross-input equality:

- type: combine
  name: fully_enriched
  input:
    orders: orders
    products: products
    categories: categories
  config:
    where: |
      orders.product_id == products.product_id
      and products.category_id == categories.category_id
    match: first
    on_miss: null_fields
    cxl: |
      emit order_id = orders.order_id
      emit product_name = products.product_name
      emit category_name = categories.name
      emit amount = orders.amount
    propagate_ck: driver

The planner builds a join tree by walking equalities pairwise: starting from the driver, it adds one input at a time, always one linked by an equality to the inputs already joined. It is designed to prefer the input with the fewest rows at each step, but those row counts are not available yet when it runs, so in practice it takes the linked inputs in the order they are declared.

Choosing the driving input

The driver is the input whose records flow through one at a time during execution; the other inputs are materialized as build-side hash tables (or IEJoin index structures). By default the first input in the input: map is the driver.

Use drive: to override:

  config:
    where: "orders.product_id == products.product_id"
    drive: products
    cxl: |
      emit product_id = products.product_id
      emit product_name = products.product_name
      emit sample_order_id = orders.order_id

With drive: products, the pipeline emits one row per product enriched with a matching order, instead of one row per order enriched with its product. Pick the driver based on which side you want to iterate over (typically the larger stream, or the one whose ordering you want to preserve).

Strategy hint

The choice is made when the pipeline is planned, from an estimate of the inputs’ size. Under auto, an equal-ids join runs in memory unless that estimate says the build side is too large. If the estimate is low or missing (for example, a glob: source whose files are not known in advance) and the build side turns out larger than the memory budget, the run stops with E310 MemoryBudgetExceeded; it does not switch to disk partway through (#1337). Set strategy: grace_hash when the build side may be larger than the memory budget: it partitions the build side to disk from the start.

Correlation-key propagation

Combine declares which correlation-key columns its output rows carry via the required propagate_ck field.

- type: combine
  name: enriched
  input:
    orders: orders
    products: products
  config:
    where: "orders.product_id == products.product_id"
    cxl: |
      emit order_id = orders.order_id
      emit product_name = products.name
    propagate_ck: driver        # driver-only (today's behavior)

    propagate_ck: all           # union of every input's $ck.* columns

    propagate_ck:
      named: [order_id]         # explicit subset (intersected with upstream)

• driver – output carries only the driver input’s correlation-key columns. Build-side records contribute body fields, but their group identity is consumed by the match; when a match fails, the build record’s dead letter follows the failing driver’s correlation group.
• all – output carries every input’s correlation-key columns. Use when the build side carries keys that downstream operators need to read.
• named: [<field>, ...] – an explicit subset. Use to project a multi-field key down to a single field after a join.

Driver wins on a name collision: if both the driver and a build input declare the same key field, the output keeps the driver’s value. See the Correlation-key combine interaction reference for how each match mode fills the propagated key (especially match: collect).

propagate_ck is required on every combine; pipelines without an explicit value fail to compile. Existing pipelines migrate by adding propagate_ck: driver, which is bit-for-bit equivalent to today’s behavior.

Output-size cap (max_output_rows)

max_output_rows is an opt-in ceiling on how many rows a combine may emit. It defaults to unlimited; set it to guard against a permissive or mis-specified predicate that would explode a small pair of inputs into a huge result (for example, a range join over an unexpectedly hot key producing a near cross product):

  config:
    where: "orders.ts >= prices.effective_from"
    cxl: |
      emit order_id = orders.order_id
      emit price = prices.amount
    propagate_ck: driver
    match: all
    max_output_rows: 1000000

Semantics:

• Fail-loud, never truncate. The moment the combine would emit more than the cap, the run stops with diagnostic E325 naming the node and the cap. It does not produce a capped or partial result — a silently truncated join would corrupt downstream data.
• Independent of the memory budget. This is a result-size guard, not memory pressure. A runaway join can be perfectly bounded in memory (its output spills to disk) yet still produce far more rows than intended; max_output_rows caps the row count regardless of bytes.
• Covers the whole output, on every strategy. The cap counts every emitted output row across all match modes and any on_miss rows, and is enforced identically whichever join strategy the planner picks (hash build-probe, grace-hash, sort-merge, or the IEJoin block-band).
• collect counts driver rows. Under match: collect a combine emits one output row per driver row (each carrying an array of up to 10 000 collected matches), so max_output_rows bounds the driver-row count, not the number of collected array elements.
• Dead-lettered rows are not counted. A matched pair whose cxl: body or residual raises a recoverable eval failure is routed to the DLQ, not the output, so it does not count toward the cap. Only that pair is dead-lettered: under match: all the driver’s other matches are still evaluated and emitted, whichever join strategy runs. A failing residual is neither a match nor a miss: under match: first a failure on the deciding candidate is the driver’s only result, and under match: collect it leaves the driver with no row (see match: first and match: collect). The driver row and the matched build row are dead-lettered, each with its own Source and row number; each failure writes its own pair, so a driver that fails against several build rows is dead-lettered once per failure, and a build row that several failing drivers matched is dead-lettered once for each of them.
• N-ary combines cap the final output. A combine whose where: spans three or more inputs is decomposed into a chain of binary steps; max_output_rows guards the final combined output, not the intermediate chain steps.

If the large result is expected, raise the cap (or omit the field). If it is not, tighten the where: predicate. Run clinker explain --code E325 for the full remediation guide.

Memory considerations

Each non-driving (build-side) input is held in memory while the join runs, so plan for roughly 1.5–2× its file size — a 50 MB lookup table needs about 75–100 MB. An equal-ids join spills its build side to disk only when it runs as a grace hash join, which the planner chooses from its size estimate or which strategy: grace_hash forces. An in-memory join whose build side outgrows the budget stops the run with E310 instead of spilling (#1337). Set the budget with pipeline.memory.limit; see Memory Tuning.

Range and equi+range predicates (the IEJoin block-band strategy) are bounded on both input axes and the output: each side is drained into disk-backed, key-sorted blocks, and the emitted rows accumulate in a spillable sort buffer rather than a resident vector. So a range join whose inputs — or whose result — exceed the memory budget spills automatically and completes, rather than failing. Even a single hot key whose block-pair is a near cross product streams through a bounded nested loop instead of materializing the whole candidate set. Use max_output_rows if you want such a runaway result to stop rather than spill.

Document boundaries

A Combine passes document boundaries through to its output, so a per-document Aggregate after a join still rolls up per document. A driver source that carries several documents (a glob: over monthly files, say) produces one roll-up per driver document after the join, not one fold spanning all of them. A document carried on both the driver and the build side opens and closes exactly once downstream. See Document Context & Envelopes for the per-document aggregation model.

Complete example

pipeline:
  name: order_enrichment

nodes:
  - type: source
    name: orders
    config:
      name: orders
      type: csv
      path: "./data/orders.csv"
      schema:
        - { name: order_id, type: string }
        - { name: product_id, type: string }
        - { name: amount, type: float }

  - type: source
    name: products
    config:
      name: products
      type: csv
      path: "./data/products.csv"
      schema:
        - { name: product_id, type: string }
        - { name: product_name, type: string }
        - { name: category, type: string }

  - type: combine
    name: enrich
    input:
      orders: orders
      products: products
    config:
      where: "orders.product_id == products.product_id"
      match: first
      on_miss: null_fields
      cxl: |
        emit order_id = orders.order_id
        emit product_id = orders.product_id
        emit product_name = products.product_name
        emit category = products.category
        emit amount = orders.amount
      propagate_ck: driver

  - type: sink
    name: result
    input: enrich
    config:
      name: result
      type: csv
      path: "./output/enriched_orders.csv"

See also

• Multi-Input Combine – recipe-style walkthrough with input data and expected output.
• Merge Nodes – streamwise concatenation; the right operator when inputs share a schema and no per-record matching is needed.
• Memory Tuning – memory budget, spill thresholds, and strategy overrides.

Aggregate Nodes

Aggregate nodes group records by one or more fields and compute summary values using CXL aggregate functions. They consume all input records in a group before emitting a single summary record per group.

Basic structure

- type: aggregate
  name: dept_totals
  input: employees
  config:
    group_by: [department]
    cxl: |
      emit total_salary = sum(salary)
      emit headcount = count(*)
      emit avg_salary = avg(salary)

Group-by fields pass through automatically – you do not need to emit them. In this example, the output records contain department, total_salary, headcount, and avg_salary.

Group-by fields

The group_by: field is a list of field names from the input schema. Records sharing the same values for all group-by fields are placed in the same group.

    group_by: [region, department]
    cxl: |
      emit total_salary = sum(salary)
      emit max_salary = max(salary)

This produces one output record per unique (region, department) combination.

Values are the same group key when they are equal under the rule sorting uses (see How values are ordered):

• Numbers group by exact value, whatever their type: 1, 1.0 and the decimal 1 are one group, and distinct integers are never merged, however large.
• -0.0 and 0.0 are one group.
• Every NaN, of either sign, is one group, separate from the group of null values.
• A group reports the value of its first-arriving row. An integer column is written as integers; a column holding both 5 and 5.0 writes 5 if the integer arrived first and 5.0 if the float did. The value is the same whether or not the Aggregate spilled to disk.

The same rule groups Cull and Reshape partition_by, a window’s group_by, correlation keys, distinct and output splitting.

Global aggregation

An empty group_by list treats the entire input as a single group, producing exactly one output record:

- type: aggregate
  name: grand_totals
  input: orders
  config:
    group_by: []
    cxl: |
      emit grand_total = sum(amount)
      emit record_count = count(*)
      emit avg_order = avg(amount)

Aggregate functions

The following aggregate functions are available in CXL:

Strategy hint

The strategy: field controls how aggregation is executed:

- type: aggregate
  name: totals
  input: sorted_data
  config:
    group_by: [account_id]
    strategy: streaming
    cxl: |
      emit total = sum(amount)

When to use streaming

The optimizer chooses streaming aggregation automatically when the Aggregate’s input is still sorted and the first fields of that order are exactly the group_by fields, in any order: take as many fields from the front of the order as there are group_by fields, and they must be the same set. Order [department, day] streams group_by: [department] and group_by: [day, department], but not group_by: [day]. An Aggregate with no group_by always streams. Use strategy: streaming as an explicit assertion – it turns a silent fallback to hash aggregation into a compile error (CXL0419), which is useful for catching sort-order regressions.

The order has to survive every stage between the Source and the Aggregate. A Merge, a Combine, a Reshape or Cull, distinct, and a Transform that writes one of the sort fields all drop it. Writing a field back unchanged counts: emit department = department drops an order on department. A Transform carries every input field through, so there is no need to emit sort fields. The sort order explainer lets you build a chain and see where the order holds.

When to use hash

Hash aggregation works on unsorted input and is the safe default. It uses more memory but handles any data ordering. Memory-aware disk spill kicks in when RSS approaches the pipeline’s memory.limit.

Correlation-key interaction

When a pipeline’s sources declare correlation_key: fields, an aggregate behaves one of two ways depending on its group_by:

• group_by covers the correlation key — if any record in a group fails, the whole group (including the aggregate output row) is sent to the DLQ.
• group_by omits a correlation-key field — only the failing records are dropped and the affected groups are recomputed, so the surviving rows still produce a correct aggregate.

You do not configure this; the engine picks the behavior from your group_by. The one restriction is that the second case cannot be combined with strategy: streaming. See Correlation Keys for the full rules.

Time-windowed aggregates

When time_window: is set on the aggregate body, records are grouped not just by group_by but also by event-time window. Each record is placed into one or more windows by its event time (the $source.event_time value derived from the source’s watermark), and each window emits one row per group once it closes.

Every upstream-reachable source must declare a watermark: so the engine knows when a window is complete. Without one, no window ever closes and the planner rejects the pipeline with E156.

The engine emits user-declared columns only — window bounds do not appear in the output unless you compute and emit them yourself. The emit order is ascending window_start (deterministic), so output rows naturally group by window.

Tumbling windows

Non-overlapping fixed-size buckets. Each record lands in exactly one window [floor(t / size) * size, floor(t / size) * size + size).

time_window:
  tumbling: { size: 1h }

Input (tumbling_demo.csv):

user_id,event_ts,kind
u1,2026-05-14T10:05:00,click
u2,2026-05-14T10:30:00,click
u1,2026-05-14T10:42:00,click
u1,2026-05-14T11:03:00,click
u2,2026-05-14T11:15:00,click
u2,2026-05-14T11:50:00,click

Output with tumbling: { size: 1h }, group_by: [user_id], emit n = count(*):

user_id,n
u1,2
u2,1
u1,1
u2,2

Reading top-to-bottom: the first two rows are the [10:00, 11:00) bucket (u1’s 10:05 and 10:42, then u2’s 10:30); the next two are the [11:00, 12:00) bucket (u1’s 11:03, then u2’s 11:15 and 11:50). Each input record contributes to exactly one window.

Hopping windows

Overlapping fixed-size buckets advanced by slide. Each record lands in ceil(size / slide) windows: slide < size produces overlap, slide == size degenerates to tumbling, slide > size produces gaps where some records fall in zero windows.

time_window:
  hopping: { size: 1h, slide: 30m }

Input (hopping_demo.csv):

user_id,event_ts,amount
u1,2026-05-14T10:05:00,10
u1,2026-05-14T10:42:00,20
u1,2026-05-14T11:10:00,15

Output with group_by: [user_id], emit total = sum(amount), emit n = count(*):

user_id,total,n
u1,10,1
u1,30,2
u1,35,2
u1,15,1

Three input records, four output rows — each record fans into two overlapping size: 1h, slide: 30m windows:

• [09:30, 10:30) — just 10:05 → total=10, n=1
• [10:00, 11:00) — 10:05 + 10:42 → total=30, n=2
• [10:30, 11:30) — 10:42 + 11:10 → total=35, n=2
• [11:00, 12:00) — just 11:10 → total=15, n=1

Session windows

Per-key gap-bounded sessions. A new record extends its key’s current session if its event time is within gap of the session’s last event time; otherwise it starts a new session. The boundary is data-driven, not clock-aligned.

time_window:
  session: { gap: 10m }

Input (session_demo.csv):

user_id,event_ts,action
u1,2026-05-14T10:00:00,login
u1,2026-05-14T10:07:00,click
u1,2026-05-14T10:13:00,click
u1,2026-05-14T10:50:00,login
u1,2026-05-14T10:55:00,click

Output with group_by: [user_id], emit n = count(*):

user_id,n
u1,3
u1,2

u1’s first three rows form one session (10:00 → 10:07 → 10:13, consecutive gaps ≤ 10m). The 37-minute idle stretch exceeds gap, so 10:50 starts a fresh session that runs through 10:55. Two sessions, two output rows.

Allowed lateness

allowed_lateness extends how long a window stays open past its end before it emits, giving late-arriving records a grace period to still be counted. It is distinct from the source-side watermark.delay. Records that arrive after a window’s end + allowed_lateness route to the DLQ as LateRecord with stage label time_window:<aggregate-name>. See DLQ category: LateRecord for the DLQ row layout.

- type: aggregate
  name: hourly
  input: clicks
  config:
    group_by: [user_id]
    time_window:
      tumbling: { size: 1h }
    allowed_lateness: 30s
    cxl: |
      emit n = count(*)

Default (unset) means no grace beyond the watermark — windows close the instant min_across_sources crosses window_end. Set allowed_lateness when the source’s watermark.delay alone is too small to absorb the observed out-of-order tail.

Worked example: multi-source session window

This pipeline merges two independent login feeds and groups per-user events into gap-bounded sessions. When several sources feed one time-windowed aggregate, a window cannot close until every source has advanced past the window’s end — the slowest source paces the others, so no window emits before all of its records have arrived.

pipeline:
  name: multi_source_session

nodes:
  - type: source
    name: src_web
    description: Web login events.
    config:
      name: src_web
      type: csv
      path: ./data/session_logins.csv
      options:
        has_header: true
      watermark:
        column: event_ts
      schema:
        - { name: user_id, type: string }
        - { name: event_ts, type: date_time }
        - { name: source, type: string }

  - type: source
    name: src_mobile
    description: Mobile login events.
    config:
      name: src_mobile
      type: csv
      path: ./data/session_mobile.csv
      options:
        has_header: true
      watermark:
        column: event_ts
      schema:
        - { name: user_id, type: string }
        - { name: event_ts, type: date_time }
        - { name: source, type: string }

  - type: merge
    name: all_logins
    inputs: [src_web, src_mobile]

  - type: aggregate
    name: user_sessions
    input: all_logins
    config:
      group_by: [user_id]
      time_window:
        session: { gap: 5m }
      allowed_lateness: 30s
      cxl: |
        emit user_id = user_id
        emit logins = count(*)

  - type: sink
    name: results
    input: user_sessions
    config:
      name: results
      type: csv
      path: ./output/multi_source_session.csv

Both sources declare their own watermark.column independently, and each source’s records carry an event time regardless of which feed delivered them — so the aggregate does not care which column name a given source used. A session cannot emit until both src_web and src_mobile have advanced past the session’s end plus its allowed_lateness. Drop the watermark: block on either source and the planner rejects the pipeline with E156.

Run it from the repo:

cargo run -p clinker -- run examples/pipelines/multi_source_session.yaml

Complete example

- type: source
  name: transactions
  config:
    name: transactions
    type: csv
    path: "./data/transactions.csv"
    schema:
      - { name: account_id, type: string }
      - { name: txn_date, type: date }
      - { name: amount, type: float }
      - { name: category, type: string }
    sort_order:
      - { field: "account_id", order: asc }

- type: aggregate
  name: account_summary
  input: transactions
  config:
    group_by: [account_id]
    strategy: streaming
    cxl: |
      emit total_amount = sum(amount)
      emit txn_count = count(*)
      emit avg_amount = avg(amount)
      emit max_amount = max(amount)
      emit categories = collect(category)

- type: sink
  name: summary_output
  input: account_summary
  config:
    name: summary_output
    type: json
    path: "./output/account_summary.json"

The output is JSON because collect(category) emits an array. JSON serializes it as a native array; XML can instead write it as repeated child elements, and CSV can join it into a delimited cell. For a scalar-only output such as fixed-width, coerce it first with a downstream Transform (for example emit categories = categories.join(";")).

Reshape Nodes

Reshape nodes observe a whole correlation group and, per group, mutate the rows whose state caused a rule to fire while synthesizing new rows derived from those trigger rows. They are the node for “look at everything an entity did, then fix one record and insert the record that should have been there” — work no other node can do:

• Aggregate reduces a group to one summary row.
• Transform emits 0 or 1 record per input record.
• Combine joins records across sources.

None of those produce new records derived from a group’s observed state while preserving the originals. Reshape does.

Reshape is a blocking grouping operator: it buffers every record of a group before any output row leaves, because a rule cannot decide what to synthesize until it has seen the whole group. It has a single output.

Basic structure

- type: reshape
  name: backfill_plans
  input: plans
  config:
    partition_by: [employee_id]
    order_by:
      - { field: plan_start, order: asc }
    rules:
      - name: fix_long_plan_years
        when: "plan_start - plan_end > 365"
        mutate:
          set:
            plan_end: "plan_start"
        synthesize:
          copy_from: trigger
          overrides:
            status: "'synthesized'"

For each employee_id group, every row where plan_start - plan_end > 365 (the trigger) has its plan_end rewritten and a new row synthesized from it with status overridden to synthesized.

partition_by

A list of field names. Records sharing the same values for all partition_by fields form one group, and every rule observes and acts within a single group. This is the correlation key the operator reasons over.

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 the rules apply across the whole dataset rather than per entity.

- type: reshape
  name: relabel
  input: rows
  config:
    partition_by: []          # one group: the whole input
    order_by:
      - { field: amount, order: asc }
    rules:
      - name: flag_large
        when: "amount > 100"
        mutate:
          set:
            label: "'large'"

order_by then sorts the whole input as a unit, and the no-cascade contract applies across every record at once. Two consequences follow from there being only one group:

• The whole input must fit memory.limit at finalize, since a single group has to be resident when its rules fire (see Limits). Whole-input grouping is for datasets that fit the budget, not for bulk row-at-a-time work — a per-record mutation with no cross-record dependency belongs in a Transform, which streams.
• A mutation conflict rolls back the whole group, so one conflict rolls back the entire run’s records rather than one entity’s.

Values Reshape cannot key

Reshape groups a record under a single null group whenever it cannot build a key from the partition value. Today that covers all of:

• the column being absent from the record
• an explicit null
• an empty string ("")
• an array- or map-valued cell (a multi-value column)

A NaN float, of either sign, is one group of its own, not part of the null group. Numbers group by exact value, as in Aggregate: 1, 1.0 and the decimal 1 are one group, and distinct integers never merge, however large.

So account="" and account=null land in the same Reshape group. Note that Cull does not fold empty strings — there, account="" and account=null are two groups. The difference is unintentional and tracked in #1022; until it is resolved, do not assume one node’s grouping matches the other’s for blank values.

If a blank-heavy column is your partition key, expect one large null group. Normalize blanks upstream (a Transform that maps "" to a real sentinel) when you want them grouped separately.

order_by

Optional. A list of sort fields applied within each group before rules run, so order-dependent synthesis is deterministic. Arrival order breaks ties.

Each entry is either a field name, which sorts ascending, or a map { field, order, null_order }:

order_by:
  - plan_start                                  # same as { field: plan_start }
  - { field: plan_end, order: desc, null_order: first }

Each group is ordered this way before its rules run, exactly as a Sink sort_order orders rows (see How values are ordered): a null goes where null_order puts it, last by default for asc and desc alike; numbers compare by their exact value across integers, floats and decimals; every NaN is one value after every number, and so comes first under desc; -0.0 equals 0.0; and rows the order calls equal keep their arrival order.

null_order: drop is rejected when the pipeline is planned. order_by only arranges the rows of a group; it never removes any. To leave out rows whose field is null, filter them in a Transform before the Reshape:

- type: transform
  name: started_only
  input: plans
  config:
    cxl: |
      filter not plan_start.is_null()

Rules

Each entry in rules: is a declarative rule with a name, a trigger predicate, and optional mutation and synthesis actions. Rules are evaluated in declaration order — but every rule observes the same original group snapshot (see No cascade below).

when — the trigger predicate

when is a CXL boolean expression evaluated against each row in the group. A row for which when is true is a trigger row for that rule: its mutate rewrites it, and its synthesize derives new rows from it. CXL boolean operators are and / or / not (Clinker’s expression language is not SQL).

mutate — in-place trigger-row mutation

mutate:
  set:
    plan_end: "plan_start"
    note: "concat(note, ' (corrected)')"

Each set: entry is field: <CXL expression>. The expression evaluates against the original trigger row and overwrites that field’s value on the row.

Two restrictions are enforced at compile time:

• A set: target must already exist in the upstream schema. Reshape mutates existing columns; it does not add new ones. Emit the column from an upstream Transform first if you need it.
• A set: may not write a partition_by field — group identity must survive Reshape.

synthesize — deriving new rows

synthesize:
  copy_from: trigger
  overrides:
    plan_date: "'2024-01-01'"
    status: "'synthesized'"

For each trigger row, synthesize emits one new row:

• copy_from: trigger — the new row starts as a copy of the trigger row’s values, then overrides are applied on top.
• copy_from: none — the new row’s columns start null; overrides must supply every column (enforced at compile time, so a synthesized row is never silently empty).

Each overrides: entry is field: <CXL expression>, evaluated against the trigger row.

Whichever copy_from a rule uses, a synthesized row keeps its trigger row’s $source.* identity — $source.name, $source.file, and $source.event_time. Downstream, $source.name on a synthesized row names the Source its trigger came from, and if a later stage dead-letters the row, the entry is attributed to that Source: it counts toward that Source’s per_source rate limit and is written to that Source’s per_source dead-letter file when one is configured. See Error Handling & DLQ.

No cascade

Every rule observes the original group state. A row mutated by rule A is not re-observed by rule B, and rule B’s when predicate sees the row’s original values, not rule A’s edits. This is a deliberate guarantee:

• Determinism — cascade would make rule order silently change output.
• Single observation — the group is observed once.

To sequence dependent transformations, chain two Reshape nodes in the DAG so the second observes the first’s output.

Mutation conflicts

If two rules write the same field on the same row, that is a mutation conflict. Some conflicts are caught at compile time when the rules’ selectors statically overlap; content-dependent collisions that cannot be proven at compile time are caught at runtime.

A runtime conflict routes a dead-letter-queue entry under the mutation_conflict category, and the whole correlation group rolls back — none of that group’s mutated or synthesized rows reach the output. The DLQ entry’s stage label is reshape:<node>:<rule_a>+<rule_b>, naming the colliding rule pair. See Error Handling & DLQ.

Known issue: today only the row where the conflict happened gets a DLQ entry. The group’s other rows are dropped without one, so they appear in neither the output nor the DLQ (#1349).

Audit stamps

Reshape stamps three engine-written columns on its output records so the provenance of a synthesized or mutated row is queryable downstream:

Like the $ck.* correlation columns, these $meta.* columns stay out of the default writer output — they are available for downstream CXL and audit, not silently dumped into your output files.

Memory model

Reshape is a blocking, grouped operator: it groups every input record by partition_by before any rule fires, because each rule must observe its whole correlation group (the no-cascade contract forbids folding a group incrementally). 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, Reshape 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, never the post-processed output rows. On reload at finalize, mutation and synthesis re-run in memory exactly as they would have without spilling, so the output is identical whether a group stayed resident or round-tripped through disk — including its within-group row order, which is restored to arrival order after a reload. (Two caveats apply; see Limits below.) Spilling input records (rather than output rows) is also what keeps a copy_from: none synthesized row — built against the wider output schema — from being reconstructed against the wrong schema; input records all share one uniform schema.
• Spill priority: 15, between grace-hash Combine (10) and external sort (20). A grouped record buffer costs more to evict than grace partitions (reload pays the re-synthesis CPU) but less than an external-sort merge. Reshape 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, Reshape 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. The cross-group resident peak is bounded near the soft limit.
• 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 so successive spill waves evict fractions of the one group — while smaller groups stay resident. The ingest-time resident peak therefore stays bounded even under one giant skewed group.

Reshape’s buffer is byte-budget-only: it does not honor the error_handling.max_group_buffer record-count cap. That cap bounds the correlation-commit group buffer used by the retraction machinery, a different buffer; Reshape’s footprint is bounded by memory.limit and the spill path, not by a per-group record count.

The on-disk spill volume Reshape 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.

Limits

Two current limitations qualify the “identical whether spilled or resident” guarantee above:

• A single correlation group must fit the memory budget at finalize. The no-cascade contract requires the whole group to be resident when its rules fire, so even though cross-group and ingest-time peaks spill to disk, the finalize reload of one group needs that group to fit. 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 Reshape node, the offending partition_by group, and its footprint against the budget:

  E310 backfill: arena exceeded budget (128000/8192) [one Reshape correlation
  group [employee_id="employee-00000"] 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, 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 Reshape’s output row is the upstream columns plus its $meta.* audit columns — it never drops a column itself — so anything you strip upstream is also missing from the written output.
  • Narrowing partition_by shrinks the group too, but that key defines the group the rules evaluate against, so a narrower key changes which records each rule sees and therefore changes your results. Treat the grouping key as a modelling decision, never as a memory knob.

  (A future two-pass finalize could lift this limit.)

  Run clinker explain --code E310 for remediation keyed to whichever memory surface overran.

• Reshape rules cannot reference $doc document context. Because the spill round-trip does not yet preserve document envelope context, a $doc.* reference in a rule’s when, mutate.set, or synthesize expression would resolve to the real envelope for a resident group but to null for a spilled one — output that depends on the memory budget. A pipeline whose Reshape rules reference $doc is rejected at compile time. Move the $doc lookup into an upstream Transform that copies the value into a record column, then reference that column in the Reshape rule.

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.
• Numbers group by exact value, as in Aggregate: 1, 1.0 and the decimal 1 are one group, and distinct integers never merge, however large.
• A NaN float, of either sign, is one group of its own, distinct from the null group.
• 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 empty strings and multi-value cells differently: it folds both into its null group instead. The blank-versus-null divergence is tracked in #1022. The multi-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 that orders the rows of each group as they are written. It does not change which groups are removed: the removal rule is evaluated over the group in arrival order (#1264). Arrival order breaks ties.

Each entry is either a field name, which sorts ascending, or a map { field, order, null_order }:

order_by:
  - txn_date                                    # same as { field: txn_date }
  - { field: amount, order: desc, null_order: first }

A group’s rows are ordered exactly as a Sink sort_order orders rows (see How values are ordered): a null goes where null_order puts it, last by default for asc and desc alike; numbers compare by their exact value across integers, floats and decimals; every NaN is one value after every number, and so comes first under desc; -0.0 equals 0.0; and rows the order calls equal keep their arrival order.

null_order: drop is rejected when the pipeline is planned. order_by only arranges the rows of a group; it never removes any, and every record of a group is kept or removed together. To leave out rows whose field is null, filter them in a Transform before the Cull:

- type: transform
  name: dated_only
  input: transactions
  config:
    cxl: |
      filter not txn_date.is_null()

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: sink
    name: kept
    input: flag_large_histories            # main port — kept groups
    config: { name: kept, type: csv, path: kept.csv }

  - type: sink
    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.

Envelope Nodes

Envelope nodes frame a body stream into per-document documents. An Envelope is a discrete, composable stage you can place after any operator — a Transform, a Merge, a Combine, or an Aggregate — to declare “from here on, treat the records as belonging to framed documents.” It mirrors the message/EDI/XML envelope-wrapper pattern (the Enterprise Integration Patterns Envelope Wrapper, XProc’s p:wrap-sequence): the body is the payload, and the envelope is the document boundary around it.

This page documents the preserve and concat framing strategies and the orthogonal header: / footer: synthesis that layers on top of either.

Interactive companion: How many documents? shows the documents a Sink writes under preserve and concat, with and without a synthesized footer, how an Aggregate after the Envelope rolls up, and which shapes E347 and E355 reject.

Basic structure

- type: envelope
  name: framed
  body: merged
  config:
    strategy: preserve

The node reads its body: input and emits the same records, framed per document. A downstream Sink with reconstruct_envelope: true then writes one framed document per body grain.

Inputs: body, the wired header, and the not-yet-wired trailer

When you omit header:, an Envelope frames each body record using the body’s own ambient envelope — the document context every record already carries from its source.

Wiring a header: port replaces the document’s header

A wired header: input is a second stream carrying one header record per document, each on the same document grain as the body it frames. The node attaches a header to a body strictly by grain, so the header record reaching it must carry the body’s grain — and the node then replaces that document’s ambient header with the wired header record (the framing grain is preserved; only the header changes). This is transform-in-place header replacement: rewrite a header’s values upstream — for example, override a batch id or stamp a run date — and frame the body with the rewritten header instead of the source’s original.

nodes:
  # `rewrite_header` rewrites the source header's values while keeping each
  # record's grain, so the rewritten header still grounds to its body document.
  - type: envelope
    name: framed
    body: payments
    header: rewrite_header
    config: { strategy: preserve }

The header record must carry a body document’s grain. A grain-preserving Transform of the source’s promoted header keeps it; a replacement from a different source establishes it via a business-key join against the body. A header record whose grain matches no in-flight body document (or carries the synthetic, ungrounded grain a Transform stamps onto a record it builds from scratch) cannot be placed, so the run fails with E351 (run clinker explain --code E351 for the full write-up):

envelope "framed": a wired header record carries document grain <grain>, which
matches no in-flight body grain (or is a synthetic / ungrounded grain). The node
attaches a header to a body strictly by grain, so it cannot place a header that
grounds to no body document.

Exactly one header record may carry each body document’s grain. When the wired header stream carries two or more records on the same grain, the node has no rule to fold a second header onto an already-framed document, so it refuses to silently keep one and drop the rest. The run fails with E352 (run clinker explain --code E352 for the full write-up):

envelope "framed": the wired header input carries two or more records for
document grain <grain> — exactly one header record per document grain is
required.

Deduplicate the header stream to one record per grain upstream — an aggregate or distinct on the grain’s business key, or a Transform that emits a single rewritten header per source document.

A wired trailer: input is still rejected this release with a clear “not yet supported” message:

envelope node "framed": explicit `trailer` input wiring is not yet supported —
omit it to frame with the body's own envelope

strategy: preserve

preserve emits one framed document per body grain. It is a transparent framing stage: body records pass through with their document context and grain unchanged, and the document-boundary signals are forwarded verbatim. Inserting a preserve Envelope between a body stage and a Sink is byte-identical to today’s per-document framing — its value is being the explicit, composable stage that later strategies extend, not a change in output.

preserve is the default, so config: { strategy: preserve } and an empty config: {} are equivalent.

Framing is keyed on the document grain, never the source file

The grain is the level at which one logical document is reconstructed — and it is not always one-per-file:

• A nested X12 interchange frames once per interchange. The GS functional-group and ST transaction-set levels inherit the interchange grain, so an ISA … IEA interchange is one framed document regardless of how many groups or transaction sets it nests.
• An HL7 multi-message file frames once per message. Each MSH message opens its own grain, so a single file containing several messages produces several framed documents.

Because framing keys on the grain rather than the source file, splitting or combining files never silently changes the document count.

strategy: concat

concat does the opposite of preserve: it collapses a multi-document body into one framed document. Every body record is re-stamped onto a single consolidated document context, so the body opens and closes exactly once regardless of how many documents fed in. This is the strategy to use when several source documents — say two files joined by a Merge — should write as one consolidated document with a single header and footer.

nodes:
  - type: merge
    name: both
    inputs: [file_a, file_b]
  - type: envelope
    name: framed
    body: both
    config: { strategy: concat }
  - type: sink
    name: out
    input: framed
    config:
      name: out
      type: csv
      path: out.csv
      reconstruct_envelope: true

Re-stamping changes only the framing (the grain) and the ambient $doc.* view a record sees — it does not disturb per-record fields. In particular $source.file is a real column stamped when each record is read, so it still reports the record’s own originating file after a concat. Concat is lossless on per-record provenance; it changes only which document the record is framed inside.

The consolidated header, and the two-headers conflict

One consolidated document can carry only one envelope header. concat derives it from the headers of the documents that contribute body records, taking one header per document:

• Every header agrees (or there is only one) → the consolidated document carries that common header.
• No document carries a header → the consolidated document is headerless.
• A headed document and a headerless document → the single header wins; the headerless document coexists with it (no conflict).

Only documents that contribute body records take part: a document that carries a header but no body records frames nothing once consolidated, so it never enters the comparison. Header identity is structural — two documents share a header when they declare the same sections, in the same order, with the same field values. Engine-added fields whose names start with $ (such as $raw, the raw segment text) are ignored, so two files whose headers differ only in raw content fold to one header, and the consolidated document keeps the first document’s full header, $raw included. A difference in a field you declare (for example an extracted control number) makes the headers distinct.

When the body carries two or more distinct non-empty headers, concat refuses to silently keep one and drop the rest. The run fails with E350 (run clinker explain --code E350 for the full write-up):

envelope "framed": concat collapses the body into one framed document, but the
body carried 2 distinct non-empty envelope headers — one document can frame only
one header, so concat will not silently drop the rest. Make the headers identical
upstream, or add a header-folding strategy that declares which header the
consolidated document keeps.

To resolve a conflict, either keep the documents separate with preserve, or make the headers identical upstream (project them to the same sections and values). header: synthesis (below) does not resolve it: concat compares the input headers and raises E350 before any synthesis runs (#1385).

Synthesizing a header and footer

header: and footer: are orthogonal to the framing strategy. The strategy decides how many output documents there are (preserve = one per body grain, concat = one consolidated); synthesis decides what header and footer each of those documents carries. The node computes a fresh header (declarative scalar expressions) and footer (streaming aggregates over the framed body) per output document, stamps them as named sections into the document’s envelope, and the same header_from_doc / footer_from_doc writer path renders them.

Both maps are keyed section -> field -> CXL expression. The inner field map preserves declaration order, which is the rendered cell order. A downstream Sink renders a section through header_from_doc / footer_from_doc, and those may name only a section that a feeding Source declares (E346), so in practice a synthesized section reuses a declared section name.

A synthesized section replaces the whole same-named section on the document; it does not add fields to it, so the section’s original fields (for example a source’s interchange.tag) are gone from that document. Other sections ride through untouched. Give header: and footer: different section names: if both name the same section, the header is applied last and replaces the footer, whose fields are lost without an error (#1388).

- type: envelope
  name: framed
  body: merged
  config:
    strategy: concat            # or preserve — synthesis works the same on either
    header:                     # section -> field -> scalar CXL
      group:                    # a different section from the footer's
        sender: $vars.sender_id
        created: $pipeline.run_date
    footer:                     # section -> field -> aggregate CXL
      interchange:
        record_count: count()
        total: sum(amount)

Header fields: evaluated once at document open

A header: field is a scalar expression evaluated once per output document, before the body streams. It may read only inputs known at document open — pipeline configuration ($vars), per-document provenance ($source), pipeline-scope state ($pipeline), and the ambient envelope ($doc.*). It may not read a body column, because there is no “current body record” when the header is emitted; a body-column reference is rejected at compile time with E353 (run clinker explain --code E353). Put body-derived values in a footer aggregate instead.

Footer fields: streaming aggregates over the body

A footer: field folds the document’s body records into a footer value at the document’s close. The fold is an O(1) accumulator per open document, so it supports exactly the streaming distributive/algebraic aggregates over a bare field or *:

Anything else is rejected at compile time. A holistic or unbounded aggregate (collect, any, weighted_avg) or a composed/multi-argument aggregate (sum(amount * 1.1), weighted_avg(value, weight)) raises E354 (run clinker explain --code E354); a non-aggregate function (median, mode) fails earlier as an unknown function. Project a composed value in an upstream Transform, or compute a holistic value in an upstream Aggregate, then aggregate the bare column here.

Synthesis is orthogonal: the same footer differs across strategies

Because the strategy sets the grain cardinality, the same footer: { interchange: { record_count: count() } } produces a different result on each strategy over the same body:

• Under strategy: preserve, each body document frames its own output document, so each footer’s record_count is that document’s body count.
• Under strategy: concat, the whole body collapses to one document, so the single footer’s record_count is the merged body count.

Placement

An Envelope is a normal single-input, single-output node — put it anywhere a record stream flows:

nodes:
  # … sources, a Combine that joins two streams into `merged` …
  - type: envelope
    name: framed
    body: merged
    config: { strategy: concat }   # preserve here is rejected with E347
  - type: sink
    name: out
    input: framed
    config:
      name: out
      type: csv
      path: out.csv
      reconstruct_envelope: true

After a Combine, Aggregate or Composition, a Sink with reconstruct_envelope: true needs a concat Envelope in between: those nodes’ rows carry no document of their own, and concat frames them as one. A preserve Envelope there is rejected with E347. For an Aggregate that is right, since its rows would pass through unframed; for a Combine the check is stricter than it needs to be, because the joined rows do keep their driver row’s document (#1384).

After a Combine this works: the joined rows keep their driver row’s document, and concat writes them as one framed document. After an Aggregate it plans without error but does not frame yet: Aggregate rows carry no source file, so the consolidated document has none either, and the Sink writes the rows with no header or footer (#603). Reshape output loses its document the same way (#1317).

Memory model

Both strategies re-park the body into the node’s own buffer slot, which the engine’s memory arbitrator governs and spills to disk under pressure — so neither strategy is bounded by total input size held in RAM.

preserve is a transparent framing pass-through: it forwards records and their document boundaries unchanged. concat additionally re-stamps each record onto the one consolidated document context and replaces the per-document boundaries with a single open/close pair; the header consolidation it does first groups the body records by document — one document’s worth of body records shares one grain and one header — so it does one pass over the records to collect the distinct headers, comparing only one envelope per document (the work is bounded by the number of documents, not the number of body rows).

header: / footer: synthesis adds, on top of the materialized body, one O(1) accumulator per footer field per open document — every allowed footer aggregate (count / sum / avg / min / max) holds a fixed-size state regardless of how many body rows it folds. So a document’s footer state is independent of its body-row count, and synthesis stays within the node’s bounded-memory model.

Sink Nodes

Sink nodes write processed records to files. They are the terminal nodes of a pipeline – every pipeline path must end at a sink (or records are silently dropped).

Terminal-node migration: type: output is retired and rejected with E376. Replace only the terminal discriminator with the paste-ready correction:

- type: sink

Composition and node output ports, produced artifacts, files and paths, serialization formats, stdout, command or machine output, writer results, and OpenLineage output datasets keep the word “output.” See Production Contracts for the compatibility boundary.

Use the same name for the Sink node and its config.name. At the current implementation boundary, mismatched names can produce a writer-mode error or empty published files instead of a planning diagnostic. The examples use matching names; planning success alone does not establish correct output for a mismatched pair.

Basic structure

- type: sink
  name: result
  input: transform_node
  config:
    name: result
    type: csv
    path: "./output/result.csv"

The type: field selects the output format: csv, json, xml, fixed_width, edifact, x12, hl7, or swift. The edifact, x12, and swift writers reconstruct one interchange/message envelope around emitted records; the hl7 writer re-emits HL7 v2 segments and optionally wraps them in batch/file envelopes. See EDIFACT Format, X12 Format, HL7 v2 Format, and SWIFT MT Format.

Structured single-writer outputs (edifact, x12, hl7, and swift) accept one concrete document grain per output file. A multi-file source or multi-input merge feeding one of these outputs is rejected instead of being silently written as one merged envelope. To write multiple structured documents, consolidate them deliberately with an Envelope node first or route each document to a separate output path.

Local and network-share destinations

An output path may be on a local filesystem or a mounted NFS/SMB share. Clinker detects the filesystem behind the actual destination and applies its contained-create and same-filesystem promotion rules there; users do not label their production paths with a CI profile. The committed filesystem matrix qualifies Clinker’s semantics against specific loopback NFSv4.1 and SMB3.1.1 mounts, but it cannot certify every vendor appliance, mount option, outage mode, or corporate network. Qualify representative production mounts before depending on atomic promotion during an outage or failover.

Clinker creates Unix output files with owner-only mode 0600. This prevents a new file from accidentally inheriting broad access in a shared drop zone. If a different service account or group must consume the result, arrange that access explicitly with the destination’s ACL/ownership policy; Clinker does not currently expose an output-mode setting.

For performance, keep spill files and optional staged input copies on a local disk when one is available. Blocking operators can create substantial random I/O, and performing that work directly on a network share adds latency and network traffic. The final output is still written as a hidden file on the destination filesystem and promoted there, so the completed file never relies on a cross-filesystem rename from local storage.

This commit lifecycle applies to single-file, per-source-file fan-out, and split: outputs. Clinker does not open or truncate an existing final while a replacement is running. Before publication, Clinker synchronizes and validates the complete output set, then promotes each hidden file directly to its final name. An overwrite is one atomic replacement rename: Clinker never moves the previous final out of the way first. It also never claims that a multi-file set can be rolled back after some replacements are already visible.

Publication is not one atomic filesystem operation for the whole set. Each individual rename is atomic, but a reader may briefly observe a mixture while the finite commit walks several destinations. If a promotion or directory sync fails, Clinker stops, exits 4, and reports three exact groups: finals that are visible and synchronized, finals that are visible but whose parent sync failed, and unpublished hidden partials. Already-visible finals stay visible; remaining old finals stay untouched. A process or machine crash in that window can leave the same mixed set plus .partial or .reservation siblings. Reconcile those named paths before retrying or consuming the set. Clinker does not create or use .backup files for output publication.

Every collision policy uses a hidden sibling reservation, including overwrite. This ensures only one live publisher may mutate a final destination. if_exists: error uses a no-replace promotion; if_exists: overwrite and clinker run --force replace only at successful promotion; if_exists: unique_suffix reserves candidate names until one wins. Reservations never expose zero-byte final placeholders. A reservation holds an operating-system lock and records its owner process. A later run reclaims it only after a short creation grace period and only when the lock is acquirable, proving that no live publisher owns it. If reservation cleanup fails after successful publication, Clinker exits 4 and names both the visible final and stale reservation as cleanup debt.

When unique_suffix can find no name at all because the destination itself refuses every candidate — the directory is not writable by this run — the diagnostic names the path you wrote, not the numbered candidate the search happened to stop on, and says that the destination rather than the name is what refused. Fix the directory’s permissions, or point path: somewhere this run may write.

Rendered fan-out paths are validated as new output paths. Directory traversal, an absolute result produced from a relative template, symbolic-link/reparse ancestors, and cross-filesystem promotion fail before a final is touched. Create the intended destination directories ahead of the run; Clinker does not follow rendered paths while creating missing fan-out parents.

{source_file} and {source_path} create one output route per discovered source file. Two source files that render to the same destination are rejected before any output is staged, with both source paths in the diagnostic. Escape a token as {{source_file}} or {{source_path}} when the braces are intended as literal filename text. Runtime source names and paths are inserted as opaque text: braces inside an actual filename are never interpreted as another token. When fan-out is combined with split:, every source has its own segment sequence: each starts at sequence 1 and rolls over independently.

With write_meta: true, Clinker writes a .meta.json sidecar for every actual committed final. A split output therefore gets one sidecar per segment, and a fan-out output gets one per rendered destination; no sidecar is written for the unrendered base template. Main outputs and sidecars share the same publication ledger, so a path collision between any two of them fails before publication and names both producers. Counters that are not known at sidecar-preparation time are omitted from the JSON rather than written as misleading zeroes.

When two paths are the same destination

Two Sink nodes — or a Sink node and a DLQ path — that resolve to one file are rejected at plan time with E322, before any record is read. Deciding that means deciding when two differently-spelled paths name one file, and that depends on the volume you are writing to, not on the text. Clinker measures the volume rather than guessing from its type, by creating and removing a probe file in the destination directory.

Two paths are the same destination when they differ only in:

• . components, or relative versus absolute spelling. ./out/errors.csv and /data/out/errors.csv from /data are one file everywhere.
• A symlinked parent directory. The existing part of each path is resolved, so a link and its target are one place.
• Letter case — only on a volume that ignores case. The default macOS (APFS) and Windows (NTFS) volumes do; ext4, xfs, and btrfs do not. Where it applies, it covers the whole of Unicode, not just ASCII: Ärger.csv and ärger.csv are one file, as are Σ.csv and σ.csv. straße.csv and strasse.csv are always two files — no filesystem treats them as one.
• Unicode normal form — only on a volume that ignores it. APFS and HFS+ do, in both their case-sensitive and case-insensitive variants; ext4 and NTFS do not. Where it applies, café.csv written with a precomposed é and the same name written as e plus a combining accent are one file. This is independent of the case rule: a case-sensitive APFS volume ignores normal form while still telling Café.csv and café.csv apart.

Because the last two depend on the volume, the same pipeline can be accepted on Linux and rejected on macOS. That is not an inconsistency — the two disks really do behave differently, and the rejection is the one that prevented a file from being written twice.

Two limits are worth knowing:

• Clinker may report a collision on a volume that would in fact have kept the two files apart — for instance on an older Windows volume whose case table predates a character you used. The run stops with both paths named, and renaming either one clears it.
• The reverse is possible on Windows, which folds some letters according to Turkish and Azeri rules that no locale-independent table reproduces. İ.csv and i.csv may be one file on such a volume while Clinker still sees two. If you write output paths that differ only in dotted or dotless I, give them distinct names.

Direct broadcast to several outputs

Several Sink nodes may name the same input. This is a broadcast: every Sink receives every upstream record, regardless of node declaration order. The run report counts one write per sink, so five input records feeding a CSV and a JSON Sink produce records_written: 10.

Use a Route node when outputs should receive different subsets. Writing a field such as _route does not select a destination; it is an ordinary output column unless a Route condition explicitly reads it.

Field control

Sink nodes can either pass every upstream field through to the writer or restrict output to the fields the upstream transform explicitly emitted. Several options control which fields appear and how they are named.

Unmapped input field passthrough

    include_unmapped: false    # Default: true

When true (the default), every field on an input record that the upstream transform did not explicitly emit still passes through to the output unchanged. This includes fields the source’s on_unmapped: auto_widen policy absorbed into the per-record $widened sidecar map – their contents expand back to top-level columns at the sink.

When false, only fields named by an emit statement in the upstream transform appear in the output. The $widened sidecar slot is stripped and undeclared input fields are dropped.

When true, how a carried-along column reaches the writer depends on the output format. Self-describing formats (JSON / NDJSON / XML) write each record’s own keys. A CSV output widens its header to the union of every record’s columns when it can materialize the batch, and otherwise — on a bounded-memory streaming path (a Merge, a fused Transform, a single-branch Route, a streaming-strategy Aggregate, or the probe side of a hash-build-probe Combine feeding the output), or an envelope-reconstructing path — fails loudly with a SchemaDrift error rather than dropping a column it cannot fit under its already-committed header. A fixed-width output has no room for an undeclared column and likewise raises SchemaDrift. See Auto-Widen & Schema Drift → Schema drift across records.

Migration notice

The default flipped from false to true in a recent release (see issue #90). Pipelines that relied on the previous behavior – where output records contained only the fields explicitly emitted upstream – must now set include_unmapped: false explicitly to restore that shape.

The flag composes independently with include_correlation_keys: true – see below. See Auto-Widen & Schema Drift -> Output controls for the full specification and cross-format flow examples.

Worked example

Suppose the upstream source emits records with order_id, customer_id, amount, and region, and a transform that emits only one derived field:

- type: transform
  name: classify
  input: orders
  config:
    cxl: |
      emit amount_bucket = if amount >= 1000 then "high" else "low"

With include_unmapped: true (the default), each output record carries order_id, customer_id, amount, region, and amount_bucket. With include_unmapped: false, each output record carries only amount_bucket. The transform’s CXL is unchanged in both cases – the Sink node decides the field set.

Include correlation-key shadow columns

    include_correlation_keys: true    # Default: false

When a source declares a correlation_key:, the engine tracks correlation-group identity on hidden columns that are stripped from output by default. Set include_correlation_keys: true to surface them in the writer output — typically for debugging correlation-group routing or auditing DLQ behavior. See Correlation Keys.

include_correlation_keys does not surface auto-widened columns – include_unmapped is the separate flag for that. The two are independent: each, both, or neither can be set.

Nested columns and writer capabilities

JSON writes map and array values recursively. XML maps ordinary keys to child elements, reserves unescaped @.../#text keys for attributes/text, and writes an array at a top-level column only when that output-facing column comes from a multiple: true declaration; arrays nested inside a map remain native XML structure. CSV likewise joins an array only for a compiled multiple: true column. An undeclared array at either writer is a routing error, not an implicit declaration. CSV, fixed-width, EDIFACT, X12, and HL7 reject a map that reaches a column slot. See JSON, XML, and Auto-Widen & Schema Drift.

Field mapping

mapping: declares the columns the file carries – which columns, under what names, in what order – without changing upstream CXL. It is a sequence, one item per output column:

    mapping:
      - order_id                  # carried through under its own name
      - sold_to: customer_id      # written as `sold_to`, read from `customer_id`
      - contact_email: customer_email
      - channel
      - sku

Two item shapes:

• A bare column name emits that column unchanged. This is the common case, and it costs one line naming the column once.
• A single-key pair renames. The output name is on the left, the source column on the right – the same side the bare form names. Reading an item left to right always tells you what appears in the file first.

The renames are the only items carrying a colon, so in a wide output they are found by scanning for structure rather than by comparing two names per line.

Order and selection

Declaration order is the output column order. Listed columns are written first, in the order the block declares them, whatever order they arrive in.

include_unmapped governs everything the block does not list. With include_unmapped: true (the default) unlisted columns are appended after the declared ones, in their existing relative order. With include_unmapped: false they are dropped, so the block becomes the complete statement of the output:

    include_unmapped: false
    mapping:
      - department
      - surname: last_name
      - first_name

Given upstream columns first_name, last_name, department, that writes exactly department,surname,first_name.

Every record carries every declared column. When a record does not supply an item’s source column, that column is still written, empty. The file’s shape follows the block, not the data — so a stream whose records differ in shape (a multi-record-type source, a column arriving through auto_widen, a composition body’s open row) still produces one stable column set in declaration order, rather than one that depends on which record happened to arrive first.

One upstream column may feed two output columns – - sku and - item_code: sku – because names must be unique on the output side, not the source side. Declaring the same output name twice is rejected (E364): a file cannot carry two columns under one header.

For the same reason, an output name that include_unmapped: true would also carry through is rejected. If upstream already has a sold_to column, writing - sold_to: customer_id under include_unmapped: true would put two sold_to columns in the file and readers would resolve the wrong one. Rename the mapped column, exclude the upstream one, or set include_unmapped: false.

Where the compiler cannot enumerate the upstream columns, the same collision reaches the run. The mapped value wins – the block is your explicit statement of what the file carries – and the displaced upstream column is named in a W366 warning at the end of the run. Applying one of the three fixes above silences it.

Diagnostics

A mapping: item naming a column that does not exist at that point in the pipeline is rejected at compile time (E365), with the available column list and a did you mean when the name is a near miss. Nothing is renamed silently.

The compiler cannot always see the column set. Inside a composition body the rows are open by construction, and under on_unmapped: auto_widen a column can reach the sink through the sidecar without being declared anywhere. There an item naming an unknown column compiles even when its name resembles a declared column: spelling similarity cannot prove that a dynamic field is absent. W365 reports it after the run if no written record supplied it.

What catches the rest is the end of the run: if no record supplied an item’s source column, that item wrote an empty column in every row, and the run reports it as W365, naming the column to correct. An item some records supply and others do not is a sparse column, not a mistake, and is not reported.

Both W365 and W366 are advisory. They print to standard error when the run finishes and do not change the exit code – the file is written and readable either way, and by the time a stream ends the run’s other outputs have already been flushed.

A column absent from the source’s schema: reaches the sink only through the auto_widen sidecar, which is expanded to top-level columns only under include_unmapped: true. A mapping: item may name such a column when that flag is set; under include_unmapped: false it cannot resolve and is rejected at compile time.

An empty block – mapping: {} or mapping: [] – is rejected (E364): it declares an output with no columns. To write every upstream column, remove the mapping: key rather than emptying it.

Writing the block as a YAML map instead of a sequence is rejected (E364); the message prints your own block already rewritten. Run clinker explain --code E364 for the migration, and read the direction note there before pasting: releases before this one documented output_name: source_field but executed the reverse, so the rewrite swaps each pair’s two sides to preserve what the pipeline was actually writing.

Excluding fields

Remove specific fields from output:

    exclude: [internal_id, _debug_flag, temp_calc]

exclude: matches incoming column names, and runs before mapping:. Two consequences:

• The columns that survive keep their relative order. Upstream a, b, c, d with exclude: [b] writes a, c, d.
• Naming a column that a mapping: item also produces is not a conflict – the exclusion removes the upstream column of that name and leaves the mapped one standing. That is the fix for the two-columns-under-one-header collision above: - sold_to: customer_id with exclude: [sold_to] writes one sold_to column, carrying customer_id’s value.

Excluding a column a mapping: item reads is a different matter, and is rejected (E364): the exclusion removes the column before the item can read it, so the item could never resolve.

Header control (CSV)

    include_header: true      # Default: true

Set to false to omit the CSV header row.

Null handling

    preserve_nulls: false     # Default: false

When false, null values are written as empty strings. When true, nulls are preserved in the output format’s native null representation (e.g., null in JSON).

Rounding decimals to a declared scale

A Sink node’s optional schema: may declare a column type: decimal with a scale. A decimal value landing in that column is rounded to the declared number of fractional places on write, using banker’s rounding — the same boundary contract a decimal source column applies on read.

    schema:
      - { name: dept,    type: string }
      - { name: total,   type: decimal, scale: 2 }
      - { name: average, type: decimal, scale: 2 }

Decimals compute at full precision inside the pipeline (division and avg keep every digit), so a declared output scale is how you pin a computed result to fixed places at the sink: avg(amount) over 1.00, 1.00, 2.00 writes 1.33 into a scale: 2 column, while sum(amount) — already at scale 2 — stays 4.00. This works for every format (CSV, JSON, fixed-width); an output column with no declared scale, or an output with no schema: block at all, keeps the full-precision value. Only decimal values in decimal-declared columns are affected — no other type is coerced. See Decimal — arithmetic rules for the full boundary-contract model.

The same rounding applies to a Sink node declared inside a composition body. When its schema: names an external .schema.yaml file, the path resolves relative to the composition file’s own directory (not the invoking pipeline’s).

Output format options

CSV

- type: sink
  name: csv_out
  input: processed
  config:
    name: csv_out
    type: csv
    path: "./output/result.csv"
    options:
      delimiter: "|"

delimiter is a single byte on the wire, so it must be exactly one ASCII character (for example ,, |, or \t). An empty, multi-character, or non-ASCII value is rejected at plan validation rather than silently truncated to its first byte.

JSON

- type: sink
  name: json_out
  input: processed
  config:
    name: json_out
    type: json
    path: "./output/result.json"
    options:
      format: ndjson           # array | ndjson
      pretty: true             # Pretty-print JSON

• array (default) – writes a single JSON array containing all records.
• ndjson – writes one JSON object per line.

JSON numbers cannot represent non-finite floats; a record carrying NaN or an infinity fails the write with a JSON error instead of silently becoming null. See JSON Format.

XML

- type: sink
  name: xml_out
  input: processed
  config:
    name: xml_out
    type: xml
    path: "./output/result.xml"
    options:
      root_element: "data"
      record_element: "row"
      attribute_prefix: "@"    # emit @-prefixed fields as XML attributes

Fields whose final path segment carries the attribute_prefix (default @, matching the XML source option) are emitted as XML attributes of their enclosing element, so attribute fields read from an XML source round-trip. See XML Format for details.

Fixed-width

- type: sink
  name: fw_out
  input: processed
  config:
    name: fw_out
    type: fixed_width
    path: "./output/result.dat"
    schema: "./schemas/output.schema.yaml"
    options:
      line_separator: crlf

Fixed-width output requires a format schema defining field positions and widths. Fields land at their declared byte ranges with gaps space-filled — see Fixed-Width Format for the layout semantics.

EDIFACT

- type: sink
  name: edi_out
  input: messages
  config:
    name: edi_out
    type: edifact
    path: "./out/result.edi"
    options:
      interchange: ["UNOA:1", "SENDER", "RECEIVER", "240101:1200", "REF1"]
      message_type: "ORDERS:D:96A:UN"
      write_una: false
      segment_newline: true

The EDIFACT writer reconstructs the interchange envelope around emitted records, recomputing the UNT/UNZ control counts and echoing the control references, and release-escapes any element data that carries a service character. The UNB header comes from interchange (literal elements) or interchange_from_doc (echoed from a $doc section). An interchange is a single envelope, so an edifact output cannot be combined with a split: block — the combination is rejected at config-validation time (E323). See EDIFACT Format for the full option reference, the record schema, and the round-trip semantics.

HL7 v2

- type: sink
  name: hl7_out
  input: messages
  config:
    name: hl7_out
    type: hl7
    path: "./out/result.hl7"
    options:
      file_header: ["^~\\&", "LAB", "HOSP", "EHR", "HOSP", "20240102", "FILE7"]
      batch_header: ["^~\\&", "LAB", "HOSP", "EHR", "HOSP", "20240102", "BATCH3"]
      segment_newline: true

The HL7 writer re-emits the MSH and body segments from the record stream, escaping any field data that carries a delimiter character (| → \F\, ^ → \S\, and so on). When a file_header (or file_header_from_doc) or batch_header is configured the writer wraps the messages in an FHS..FTS file or BHS..BTS batch and recomputes the closing BTS/FTS counts. A batch/file envelope is a single structure, so an hl7 output cannot be combined with a split: block — the combination is rejected at config-validation time (E339). See HL7 v2 Format for the full option reference, the record schema, the MSH off-by-one, and the round-trip semantics.

Sort order

Sort records before writing:

    sort_order:
      - { field: "name", order: asc }
      - { field: "amount", order: desc, null_order: last }

• first – nulls sort before all non-null values.
• last – nulls sort after all non-null values.
• drop – records with null sort keys are excluded from output.

drop is available only on a Sink sort_order, because only a Sink’s ordering decides which records are written. A Source sort_order, a Cull or Reshape order_by and a Transform analytic_window.sort_by only order records, so they accept first and last and reject drop when the pipeline is planned, pointing at a filter not <field>.is_null() Transform instead. When CXL cannot name the field as it is (a name with a space, a CXL keyword such as filter, or a flattened Address.City), the error prints no CXL and points at the Source schema’s source_name rename, which gives the column a name the filter can use.

drop removes records, so a run using it writes fewer records than it read and that is not a fault. A missing column counts as a null key: a record that never carried the sort field is dropped the same as one carrying an explicit null. With several dropping fields, a record is excluded if any of its keys is null, and counts once however many of them are.

The excluded records are counted, separately from records_dlq and from filter losses, so a short output can be attributed rather than guessed at. A run that dropped any reports the number on completion:

1234 record(s) excluded by null_order: drop

and the same number is written as records_null_dropped in the metrics spool when one is configured (see Metrics).

Under fan-out the count is per exclusion, not per source record: two Sinks that each declare a dropping sort_order each drop their own copy, so one source record excluded at both counts twice — the same multiplicity records_written carries. Subtracting this from records_total is therefore only sound on a pipeline with a single dropping Sink.

Nothing else records these records. Unlike a DLQ entry, a dropped record leaves no artifact to inspect afterwards – if you need to see which records were removed rather than only how many, route them out with a filter before the sort instead of declaring drop.

Shorthand: a bare string defaults to ascending with nulls last:

    sort_order:
      - "name"
      - { field: "amount", order: desc }

A Sink sort_order materializes all records that reach that terminal and re-establishes one order across them, including records from several physical files or Merge inputs. The guarantee is exactly the authored field sequence, direction, and null placement. drop is also part of the authored contract: records with a null in a sort key do not reach the writer.

For a split Sink, that order is global across the complete numbered split set, not restarted independently inside each file. Clinker sorts and applies null_order: drop before it rotates the writer. Each numbered file is therefore a contiguous slice of the one ordered sequence, and concatenating the files in sequence-number order recovers that sequence. Dropped rows do not count toward max_records, max_bytes, or the resulting number of split files.

The sort is stable. Equal authored keys retain their upstream arrival order within a given execution path, and the same path produces the same bytes in resident and forced-spill operation. Clinker does not add a source-row, filename, or canonical-record tie-breaker. If upstream strategies can produce different arrival orders, equal-key rows have no cross-strategy relative-order promise. Author enough fields for a total business order before using an exact byte comparison; otherwise validate the decoded record multiset and aggregate values instead.

How values are ordered

Every sort uses one rule for comparing two values: a Sink or Source sort_order, a Cull or Reshape order_by, a window’s sort_by, and the check that verifies a Source’s declared order. The rule does not depend on the memory limit, so a sort that spills to disk writes the same records in the same order as one that fits in memory.

• Nulls are placed only by null_order: first, last or dropped. A missing column counts as a null.
• Values of one type order naturally: numbers by value, strings by UTF-8 code point (no locale collation, so "Z" sorts before "a"), false before true, and dates and datetimes chronologically. A leap-second datetime sorts with the instant one second later that has the same fraction.
• Integers, floats and decimals compare by their exact value, not through a rounded floating-point copy. The integer 1, the float 1.0 and the decimal 1.00 are equal. The integer 9007199254740993 sorts after the float 9007199254740992.0, although the two round to the same float. The decimal 0.1 sorts before the float 0.1, whose exact binary value is slightly larger.
• Zero has one position: -0.0 and 0.0 are equal.
• NaN is one value, whatever its sign. It sorts after every number, inf included, in ascending order, and so comes first in descending order.
• Values of different types, which a column can hold when an expression’s branches produce different types, order by type: booleans, then numbers, then strings, then dates, then datetimes, then arrays, then maps.

Values the rule calls equal keep their arrival order, as described above, at every memory limit.

Physical writer boundaries

Planning derives the writer boundary from the finalized graph, not from how many Sink nodes appear in the YAML. The same ordering promise is therefore enforced at every physical byte-emission path:

• ordinary single-file and split-file record output;
• one output per physical source file;
• reconstructed envelope output per document;
• document DLQ output after the whole document is known to be clean;
• deferred output per correlation group; and
• incremental streaming output.

Complete-population modes apply the exact authored key at their population boundary using the same bounded-memory spill path. Incremental streaming cannot truthfully promise a terminal whole-population sort. If a finalized output mode is incompatible with an authored sort_order, planning rejects the pipeline instead of weakening the promise. The diagnostic names the Sink, mode, authored keys, and last reordering stage, and includes a corrected sort_order form that can be pasted into the source or upstream node.

File splitting

Split output into multiple files based on record count, byte size, or group boundaries:

- type: sink
  name: split_output
  input: processed
  config:
    name: split_output
    type: csv
    path: "./output/result.csv"
    split:
      max_records: 10000
      max_bytes: 10485760           # 10 MB
      group_key: "department"       # Never split mid-group
      naming: "{stem}_{seq:04}.{ext}"
      repeat_header: true           # Repeat CSV header in each file
      oversize_group: warn          # warn | error | allow

Split configuration fields

At least one of max_records or max_bytes should be specified for splitting to have any effect.

The naming grammar is strict: {stem}, {ext}, and the one required {seq:NN} token are the only placeholders. Unknown placeholders, a bare {seq}, non-numeric or out-of-range widths, and duplicate or missing sequence tokens are rejected during configuration validation. For example, {stem}_{seq:03}.{ext} renders sequence 7 as 007.

For formats whose output wraps the whole file in framing – a JSON array or an XML root element – each split file is a complete, independently valid document: the framing is closed at rotation and reopened for the next file.

When the Sink also declares sort_order, splitting happens after the complete Sink population has been ordered and null-key drops have been applied. Segment 1 receives the first surviving records, segment 2 the next records, and so on. The files are individually ordered and together form one ordered sequence when read by sequence number; split rotation never starts a new independent sort.

Oversize group policies

• warn (default) – log a warning and allow the oversized file.
• error – stop the pipeline.
• allow – silently allow the oversized file.

When group_key is set, the split point is the first group boundary after the threshold is reached (greedy). Without group_key, files are split at the exact limit.

Streaming writes after an interleave Merge

When a single Sink sits directly after a Merge with mode: interleave whose inputs are all Sources, records are written to disk as they arrive rather than being buffered until the merge finishes. This keeps memory flat and lets a slow writer naturally pace the upstream readers.

- type: source
  name: src_a
  config: { type: csv, path: a.csv, schema: ... }
- type: source
  name: src_b
  config: { type: csv, path: b.csv, schema: ... }
- type: merge
  name: merged
  inputs: [src_a, src_b]
  config:
    mode: interleave        # required
- type: sink
  name: out
  input: merged
  config:
    name: out
    type: csv
    path: out.csv

This is automatic — there is no setting to enable it. It applies only to this exact shape: one interleave Merge of Sources feeding one non-splitting Sink, in a pipeline without correlation keys. Any other topology buffers as usual. Both paths preserve the same record multiset and writer semantics, but an unseeded interleave does not promise one exact cross-input row sequence. Add a Sink sort_order with a total business key when exact bytes are required.

Complete example

- type: sink
  name: department_reports
  input: enriched_employees
  config:
    name: department_reports
    type: csv
    path: "./output/employees.csv"
    # `include_unmapped: false` makes the mapping the whole output: these four
    # columns, in this order, and nothing else. Without it every unlisted
    # upstream column would still be appended after them, and an `exclude:`
    # would be needed to keep any of them out.
    include_unmapped: false
    mapping:
      - "Employee ID": employee_id
      - "Full Name": display_name
      - department
      - "Annual Salary": salary
    include_header: true
    sort_order:
      - { field: "department", order: asc }
      - { field: "display_name", order: asc }
    split:
      max_records: 5000
      group_key: "department"
      naming: "employees_{seq:03}.csv"
      repeat_header: true

CSV Format

CSV is the default file format. The reader decodes each CSV row (including quoted multiline cells) into a record whose fields are matched positionally (or by header name) against the source’s declared schema:; the writer reverses the process. CSV pairs with the file transport — see Source Nodes for the transport / format split and the schema rules every source shares.

- type: source
  name: orders
  config:
    name: orders
    type: csv
    path: "./data/orders.csv"
    schema:
      - { name: order_id, type: int }
      - { name: customer_id, type: int }
      - { name: amount, type: float }
      - { name: order_date, type: date }
    options:
      delimiter: ","         # default ","
      quote_char: "\""       # default "\""
      has_header: true        # default true
      encoding: "utf-8"      # default "utf-8"

Declared types

CSV has no native numeric or date types: decoding first produces text cells. Source ingestion then parses and validates each declared column against schema: before buffering, sorting, or downstream CXL evaluation. With type: int, the cell 42 reaches a Transform as an integer; with type: string, the cell 0042 remains text, including its leading zeroes. No explicit CXL cast is needed for a column already declared with its intended type. Casts such as .to_int() are useful when a pipeline deliberately keeps text in its schema and converts it later.

A value that fails its declared type rejects the row under the configured error policy; it never silently falls back to text. See Declared-type failures.

Options

All CSV options are optional. With no options: block, Clinker uses standard RFC 4180 defaults.

delimiter and quote_char are each a single byte on the wire, so each must be exactly one ASCII character. An empty, multi-character, or non-ASCII value (for example "||" or "→") is rejected at plan validation — it is never silently truncated to its first byte.

Encoding

The reader decodes every field through the source’s declared encoding:

• utf-8 (the default) is strict — a byte sequence that is not valid UTF-8 fails the run loudly rather than substituting replacement characters, so a mis-declared encoding is caught instead of silently corrupting data.
• iso-8859-1 (Latin-1; also spelled latin-1 or latin1) maps each byte 0xNN to codepoint U+00NN, so high bytes such as 0xE9 (é) from legacy exports decode correctly.

The same options.encoding setting is available on a CSV Sink. It applies to headers, body fields, joined values, embedded JSON cells and reconstructed envelope rows. UTF-8 is the default on both sides. Names are case-insensitive; hyphens, underscores and spaces are ignored. UTF8, ISO8859_1, latin1, Latin-1 and l1 resolve to the corresponding canonical spelling.

Latin-1 is true ISO-8859-1: bytes 0x80..0x9F are the corresponding control codepoints, not punctuation from another code page. Characters above U+00FF (for example €) cannot be written in Latin-1 and cause an error; there is no replacement character or fallback. UTF-8 input strips its leading UTF-8 BOM; Latin-1 preserves an initial EF BB BF sequence as the three ordinary characters . CSV grammar is parsed from bytes before either header or body text is decoded.

An unsupported encoding is rejected during configuration admission with a precise error naming the value and a supported correction. Charset is part of the semantic plan identity; changing input or output charset changes that identity. Aliases and an omitted or explicit UTF-8 default resolve identically. Malformed UTF-8 in a header or body cell is an input-data error (source.data.invalid), including when the header is read to discover columns.

The closed encoding policy across formats is:

Single-schema and multi-record CSV sources support the same two encodings, including textual column headers, discriminator fields, body cells and declared envelope sections. Each file applies its own leading-BOM rule. A UTF-8 BOM is recognized only at the start of that file; the same bytes inside a field are data.

Header handling

With has_header: true, the header row’s names bind input columns to the schema: entries — column order in the file may differ from the schema. With has_header: false, binding is strictly positional, so the schema order must match the file’s column order.

Input columns the schema does not name are governed by the source’s on_unmapped policy, the same as every other format.

Multi-value cells (split_values)

A CSV cell holds one string, but that string may pack several values behind a delimiter (1,a;b;c). Declare the column multiple: true and add a split_values entry naming the field and its delimiter, and the reader parses the cell into an array:

- type: source
  name: orders
  config:
    name: orders
    type: csv
    path: ./orders.csv
    split_values:
      - { field: tags, delimiter: ";" }
    schema:
      - { name: order_id, type: string }
      - { name: tags, type: string, multiple: true }

tags reads as ["a", "b", "c"]. An empty cell is an empty array; a cell with no delimiter is a one-element array; each element is coerced to the column’s declared type:. A quoted cell is unquoted first, so a delimiter inside the quotes is not a boundary. A multiple: true column with no covering split_values entry is rejected at compile (E361). Multi-record CSV sources reject split_values and multiple: true columns with E358 and E361; these input options require a single-schema source. See split_values in the Source reference for the full grammar.

Writing CSV

On output, the writer emits one row per record with cells in the output schema’s column order — the same order as the header row — regardless of how an upstream node ordered the record’s fields. An output-schema column the record does not carry emits an empty cell, the same as an explicit null; with include_unmapped: false a record field the output schema does not name is not written. See Sink Nodes for header control, field mapping, and null handling.

- type: sink
  name: export
  input: orders
  config:
    name: export
    type: csv
    path: ./out/orders.csv
    options:
      encoding: iso-8859-1

Each output operation is prepared completely before any of its bytes reach the destination. The first body row and its automatic header are one operation. An unrepresentable cell therefore writes neither a partial row nor a stray header, and leaves previously accepted bytes and format state intact. Explicit document start and end operations are prepared separately. An I/O failure during delivery can leave a destination prefix and prevents further writes; successful preparation alone is not a delivered row. File publication is a separate contract described in Storage & Spill Location.

CSV quoting needs each complete cell. Its encoding workspace is admitted against the run’s resource budget before allocation and released after that cell; there is no authored field-size or record-size ceiling. A cell that cannot fit fails as a resource error rather than being truncated or dropped. Joined or embedded JSON cells account for the rendered text and encoded bytes while both are live. The complete operation also needs storage for its prepared bytes: it stays in memory unless an explicit spill location is available. Spill does not eliminate the minimum memory needed for a cell, policy, or schema mapping. See Memory Tuning for the accounting boundary and output preparation for storage and cleanup behavior.

Split output captures only the header actually delivered by a successful operation. With repeat_header: true, later files replay those same names; with repeat_header: false, only the first file emits the automatic header. include_header: false suppresses that header throughout. Reconstructed envelope rows do not become an automatic column header.

An empty stream emits no automatic header. A single empty or null cell is written as "" followed by a newline; adjacent empty cells are separated by the delimiter. Reading an ordinary empty cell yields an empty string, so CSV does not distinguish an empty string from a null unless the pipeline applies its own schema policy.

Writing multi-value cells (join_values)

A multiple: field is joined into one delimited cell on write — the write-side inverse of split_values. The default needs no configuration: values join with ;, and a value that itself contains the delimiter is a hard error rather than a cell that would split back wrongly. The planner carries the exact output-facing multiple: true column set through mapping and exclusion into the writer. An array reaching any other CSV column is rejected as a routing/type-contract error rather than joined implicitly.

- type: sink
  name: report
  input: orders
  config:
    name: report
    type: csv
    path: ./out/report.csv
    join_values:
      - tags                              # delimiter ";", on_conflict: error
      - { field: notes, delimiter: "|", on_conflict: escape, escape: "\\" }

A field with no join_values entry still joins, with the defaults. An entry overrides, per field:

• delimiter — the separator written between values (default ;).
• on_conflict — what to do when a value contains the delimiter:
  • error (default) — reject the record with the field and element position, preserving its original value for the DLQ, rather than emit a cell that splits back wrongly. This is what makes a defaulted delimiter safe. Under error_handling.strategy: continue, the offending record goes to the dead-letter queue (category multi_value_join_collision) and the run continues; under fail_fast it aborts. The exception is a pipeline where any Source declares dlq_granularity: document: there the collision fails the run (see Not covered, #933).
  • escape — prefix each delimiter (and each escape character) inside a value with escape (default \), so a matching split_values escape: recovers the original. Lossless. delimiter and escape must each be a single character.
  • encode_json — encode the whole field as an embedded JSON array, recovered by a matching split_values json: true. Preserves every value’s text exactly, including ones carrying the delimiter, quotes, or newlines — nothing is lost or mis-split. (A decimal/date/datetime element serializes as its JSON string form and reads back as a string, re-typed by the column’s declared type:, the same round trip every CSV cell takes.)

An empty field emits an empty cell — and, under the delimited policies (error, escape), a single empty-string value [""] emits an empty cell too, which reads back as zero values: the delimited encoding cannot tell an empty field from one empty value. Use encode_json when that distinction matters. A single non-empty value emits that value with no delimiter. The joined cell is quoted by the normal CSV rules when it contains the field delimiter, a quote, or a newline. Declaring join_values on a non-CSV output is rejected at compile (E362).

Round trip. on_conflict: escape and encode_json are recovered exactly by a matching source split_values entry:

# write side
join_values:
  - { field: tags, on_conflict: escape, escape: "\\" }
# read side (a later pipeline)
split_values:
  - { field: tags, escape: "\\" }

Header widening under auto-widen

When auto_widen is in effect and the Sink leaves include_unmapped at its default of true, different records can carry different carried-along columns. The header must still be shared by every row, so Clinker widens it to the union of every record’s columns in first-seen order: a column that first appears on a later record still gets its own header slot, and the earlier rows write an empty cell for it. This pre-scan runs on the buffered output path, where the record batch is materialized.

An output that streams under a bounded-memory budget cannot pre-scan the whole batch: a CSV output fused directly after a Merge/Transform, a single-branch Route, a streaming-strategy Aggregate, or the probe side of a hash-build-probe Combine, or one reconstructing an envelope (which suppresses the shared header and streams a headerless body), commits its columns to the first record. A later record carrying a column that first record lacked then fails the run with a SchemaDrift error naming the column, rather than silently dropping it. Declare the column in the source or output schema: so every record carries it, or route to a self-describing format (JSON / NDJSON / XML). A reconstruct_envelope CSV output therefore requires a stable body shape — every record must carry the same columns.

Multi-record files (header / trailer / body)

Some CSV exports interleave multiple record types in one file — a header row, many body rows, and a trailer row — each distinguished by a discriminator column. Declare these with a map-form schema: carrying a discriminator: and a records: list, instead of the single column-list schema:. Each record type names its tag (the discriminator value that identifies it) and its own columns:; the discriminator field must sit at the same column in every type (usually the first). The reader derives the runtime superset schema (a lead record_type column plus the union of every record type’s columns) automatically.

- type: source
  name: payments
  config:
    name: payments
    type: csv
    path: "./data/payments.csv"
    schema:                                     # one multi-record schema (map form)
      discriminator: { field: rec_type }        # the physical column carrying the type tag
      records:
        - { id: header,  tag: H, columns: [ { name: rec_type, type: string }, { name: batch_id, type: string } ] }
        - { id: detail,  tag: D, columns: [ { name: rec_type, type: string }, { name: id, type: int }, { name: amount, type: int } ] }
        - { id: trailer, tag: T, columns: [ { name: rec_type, type: string }, { name: count, type: int } ] }
      structure:
        - { record: trailer, count: count }     # validate T's count against the body count
    envelope:
      sections:
        head:
          extract: { record_type: H }          # the H record type surfaces as $doc.head.*
          fields:
            batch_id: string

The reader emits one record per CSV row, including quoted multiline cells, on a single superset schema whose lead record_type column carries the matched type’s id. A downstream Route discriminates on that column. Rows of different record types may carry different column counts (ragged rows) — the reader validates the column count per record type, not file-wide. A textual column-header row is skipped when has_header is true (the default), so a leading record_type,name,amount line is not mistaken for a record of an unknown type. Each declared field honors its own type / trim / pad, the same as a single-record CSV field.

• Header rows declared as an envelope: section via the record_type extract surface as $doc.<section>.* and are excluded from the body stream (see Envelopes & Document Context).
• Trailer rows named by a structure: constraint are validated as they stream — the declared count field is checked against the actual body-record count at document close — and excluded from the body stream. A declared trailer that never appears is an incomplete-document error; a body row after the trailer is rejected as content past the document close.
• Blank lines (empty or whitespace-only, common after concatenation) are skipped rather than parsed.
• An unknown discriminator value (a tag no records: entry declares) is a structural-integrity failure, classified separately from a trailer count mismatch. It aborts the run under fail_fast; under continue with the default record granularity it dead-letters only that physical row and continues with the next row; under dlq_granularity: document it condemns the whole file. A record-grained DLQ row carries a JSON array of the decoded CSV cells in _cxl_dlq_source_record, preserving empty cells without guessing which declared layout the unknown tag meant.

JSON Format

The JSON reader turns a JSON document into a record stream. It handles three physical shapes — a single array of objects, newline-delimited objects (NDJSON), or a wrapper object that nests the records under a path — and auto-detects the shape when you do not declare it. Each object is matched against the source’s declared schema:; see Source Nodes for the shared schema and transport rules.

- type: source
  name: events
  config:
    name: events
    type: json
    path: "./data/events.json"
    schema:
      - { name: event_id, type: string }
      - { name: timestamp, type: date_time }
      - { name: payload, type: string }
    options:
      format: object          # array | ndjson | object (auto-detect if omitted)
      record_path: "data"     # dot-separated keys to the records array
      max_index_bytes: 64MB   # cap on retained envelope sections (optional)

Text encoding

Input is strict UTF-8. One leading UTF-8 BOM is accepted and removed at each physical file open, including an envelope pre-scan. UTF-16 and UTF-32 BOMs and malformed UTF-8 are rejected; convert the file to UTF-8 before running it. There is no JSON encoding option or lossy fallback. A later invalid file does not erase records already delivered from earlier files. Validation follows the reader: it does not promise to discover malformed bytes beyond what it reads.

Output is UTF-8 without a BOM. Ordinary format: ndjson always writes one compact object followed by exactly one LF, including the last record; pretty: true does not expand ordinary NDJSON across lines. pretty still controls array output and reconstructed envelope documents. Envelope framing is documented separately.

Physical shapes

If format is omitted, Clinker auto-detects the shape from the file content. Declare it explicitly when the file is large enough that you want to skip detection, or when an object wrapper needs a record_path.

record_path

record_path is a dot-separated path of object keys, descended from the document root. data.rows selects the array at {"data": {"rows": [ … ]}}, and each of its elements becomes one record. This is the canonical statement of the grammar; other pages link here rather than restate it.

The rules, in full:

• No $. root marker. It is not JSONPath. Write data.rows, not $.data.rows. Only the exact leading $. is rejected, so a key that merely starts with $ ($schema.rows) is still addressable.
• No leading /. A leading slash is how a JSON Pointer is anchored; record_path is already anchored at the document root.
• No empty segments — no doubled separator (data..rows) and no trailing one (data.).
• Omitting record_path entirely lets the reader auto-detect the document shape. That is not the same as record_path: "", which is a path naming a key called “” and is rejected.

A value breaking any of these fails at compile time with E363, before any input is opened. The diagnostic names the corrected path where one can be derived.

record_path takes precedence over format:. When both are declared the reader navigates the path and streams the array it finds, whatever format: says — so pair record_path with format: object (or leave format: off). Declaring format: ndjson alongside a record_path does not read NDJSON.

Because a JSON key may contain any character, the two rejected prefixes give up a sliver of addressing: a top-level key literally named $ followed by a nested key, and a top-level key whose name starts with /, are not reachable through record_path.

Nested arrays

JSON records frequently embed arrays — line items on an invoice, tags on a product. Three source-level declarations decide what happens to them, all documented on the Source Nodes page:

• split_to_rows fans the array out to one record per element. mode: extract (the default) hoists an object element’s keys onto the output record; mode: split keeps the record shape, flattening the element back under the field name (orders.id). An array of scalars keeps the value under the field’s own name under both modes.
• A schema column declared multiple: true keeps the array as an array, and normalizes a lone scalar into a one-element array so the column’s shape never depends on what a particular document happened to carry.
• split_values parses a delimited string cell into several values.

A record whose declared field holds an empty array, is explicitly null, or carries no such field at all, is preserved by default — keep_empty defaults to true, and setting it to false drops such a record. An explicit null is how many producers write “no value”, so it counts as no occurrence rather than one; for the same reason a multiple: true column holding an explicit null stays null rather than becoming [null].

A field that IS present but holds a single object or scalar rather than an array is one occurrence, projected exactly as a one-element array would be. Producers routinely unwrap a lone element, so a feed where some documents carry "line_items": [{…}, {…}] and others carry "line_items": {…} fans both out the same way and every output record ends up with the same columns. The XML reader, where a document cannot express the difference at all, already behaved this way.

Two declared fan-out fields apply in declaration order and multiply. A nested pair (orders then orders.items) produces the two-level expansion when the outer entry declares mode: split:

    split_to_rows:
      - { field: orders, mode: split }
      - { field: orders.items, mode: split }

Under mode: extract the outer entry lifts the occurrence’s keys to the top level, which removes the orders.items path the inner entry addresses — so that pairing is rejected at compile (E358) rather than silently fanning out only one level. A duplicated field is rejected too.

Set source-level max_output_rows_per_input: N to bound the cumulative product without materializing it. The reader emits the first N rows in stable order, then a first attempted row above the ceiling routes the original JSON object to the DLQ as expansion_limit_exceeded; 0 or omission is unlimited. See Source Nodes -> split_to_rows for the complete error and fail_fast behavior.

Flattened-name collisions

The reader dissolves nested objects into dotted keys, so {"a": {"b": 1}} becomes the field a.b. When two distinct keys flatten to the same name — for example a nested {"a": {"b": 1}} alongside a literal {"a.b": 2} in the same record — only one value could survive, and keeping one while dropping the other is silent data loss. The reader refuses the record instead, naming the colliding field. This mirrors the XML reader’s treatment of a repeated element: both formats now fail loud on an undeclared collision rather than one keeping the first value and the other the last. If the collision is intentional (both values belong together), declare the column multiple: true to collect them into an array in document order; otherwise rename one of the source keys so they no longer collide. As with XML, detection is per document at read time, so the run aborts under fail_fast and dead-letters the document under continue with dlq_granularity: document.

Detection covers two distinct source keys that flatten to the same dotted name — the nested {"a": {"b": 1}} plus literal {"a.b": 2} case above. It does not cover a key that is literally duplicated within one JSON object ({"tags": "x", "tags": "y"}): the JSON parser collapses such duplicates last-wins (keeping "y") before the record reaches collision detection, so that repeat is silently dropped rather than reported. A collision inside an array element that a split_to_rows: extract fan-out lifts to the top level (one element key clashing with a parent field or with another element key) is likewise not yet detected and still resolves last-wins — tracked by issue 920.

Bounding envelope retention: max_index_bytes

When a source declares an envelope: and a pipeline reads $doc.* paths from it, the JSON reader runs a streaming pre-scan that walks the document once and retains only the declared section subtrees — every other key, including a multi-megabyte body array, is parsed-and-skipped without being stored. The retained sections live in a bounded document index.

max_index_bytes caps that index. It is charged incrementally as each section is parsed, so even a single oversized declared section aborts mid-parse (naming the section and the cap) rather than risking an out-of-memory failure. It accepts a decimal size string (64MB, 500KB) or a bare byte count; optional, defaulting to 64MB. Only the declared sections a program actually reads are retained, so envelope metadata sits far below this ceiling in practice — the cap exists to convert an unbounded mistake into a clear error. See Document Envelope Context for the full model.

Non-finite floats

JSON numbers cannot represent NaN, +infinity, or -infinity. Writing a record (or an envelope section field) that holds a non-finite float to a JSON output fails with a bounded field diagnostic, rather than silently substituting null — a substituted null would be indistinguishable from a genuine source null on read-back. Filter such records or replace the value in a transform before the JSON output.

Writing JSON

A JSON output writes one object per record, in schema-column order, either as a single array (format: array, the default) or one object per line (format: ndjson).

- type: sink
  name: enriched
  input: processed
  config:
    name: enriched
    type: json
    path: "./output/enriched.json"
    preserve_nulls: false  # omit null columns; native map/array nulls remain values
    options:
      format: ndjson     # array | ndjson
      pretty: false      # indentation for arrays or reconstructed envelopes

Dotted column names become nested objects

A column name containing a . expands back into nesting, the same way the XML writer expands one into nested elements. Columns Address.City and Address.State write as one object:

{"Address":{"City":"Boston","State":"MA"},"name":"Ada"}

This is what makes a JSON-in / JSON-out pipeline reproduce its input shape: the reader flattened {"Address":{"City":…}} into the column Address.City, and the writer puts it back. It applies to every JSON output, with no option to turn it off — a flag would mean the same column name meant different things at different outputs.

Three points follow from the rule, all shared with the XML writer and specified in full on Field Paths:

• Grouping. Columns sharing a prefix collect into one object, positioned where that prefix first appeared, even when the schema interleaves them.
• Absent children. Under preserve_nulls: false a null column emits no key, and an object whose every descendant is absent emits no key at all rather than an empty {} — so it reads back as the absent column it stands for.
• Values are untouched. A column holding a map or an array still serializes as that map or array. Expansion adds structure above the value, never inside it.

Native map and array values

CXL can construct maps, arrays, and array comprehensions directly; a JSON output writes those values recursively as native objects and arrays. Map key insertion order and array item order are preserved:

emit payload = {
  customer: customer_name,
  items: [{sku: item.sku, quantity: item.quantity} for item in line_items],
}

The output contains "payload" as an object with an "items" array. Native maps and arrays are values, so the default preserve_nulls: false does not remove null map entries or array items inside them; it controls null output columns and null leaves created by dotted-column expansion.

JSON and XML share one neutral-map key grammar. After CXL has decoded the string literal, an unescaped key is its ordinary logical spelling. Exactly one leading backslash marks a literal reserved-looking key only in these three forms: \@name, \#text, or \\name. The neutral decoder removes that one marker. In CXL source, where the string literal itself must escape the backslash, write "\\@name", "\\#text", or "\\\\name". Other leading-backslash forms are non-canonical and fail. JSON assigns no structural role to @name or #text, but it still uses this same decoder: "\\@literal" writes the JSON key "@literal".

Static and computed map keys follow the same rule. Two authored spellings that decode to the same logical key are a duplicate and fail rather than selecting a winner. Before writing any bytes for a record, the writer validates the entire neutral tree. Scalars have depth zero and each map or array adds one container; depth 64 is accepted and depth 65 is rejected. A failed nested value therefore cannot leave a partial JSON record in the output.

This recursive behavior is native to JSON/NDJSON and XML. Flat, positional, and message formats do not silently turn a map or array into JSON text. Use an explicit encoding the destination format declares—such as join_values for a multi-value flat field—or reshape the value before that output. Without one, the structured value is rejected before bytes for that record are written.

Keeping a literal . in a key

To emit a key that genuinely contains a ., escape the separator in the column name. The column a\.b writes the single key "a.b":

      schema:
        - { name: "a\\.b", type: string }   # emits {"a.b": …}
        - { name: "a.b",   type: string }   # emits {"a": {"b": …}}

A [ in a column name is currently literal but reserved; write \[ if you want it to stay literal indefinitely. See Field Paths for why.

Note that this is a write-side escape. A source key that literally contains a . still arrives from the reader as an unescaped column name (the Flattened-name collisions section above covers what the reader does), so {"a.b": 1} read and written back comes out as {"a": {"b": 1}}. Closing that is tracked by issue 920.

Column names that cannot both be written

Two columns can describe places that cannot both exist in one object — a column a holding a value alongside a column a.b that needs a to be an object. Rather than keep one and drop the other, the writer refuses the whole column set before emitting that record, identifying the offending column and the path rule to correct. Field Paths lists every clashing shape.

A column name carrying a malformed escape — a \ that is not part of \., \[, or \\, as in a column literally named C:\temp — is refused the same way, with escape guidance; write C:\\temp for a literal backslash.

Preparation and empty output

Each complete output operation is prepared within the run’s finite resources before delivery. An invalid value or resource refusal during preparation writes none of that operation. A destination failure during delivery can leave a prefix; the writer then stops and never retries or finalizes on teardown. Earlier delivered records remain delivered. See output preparation for spill, cancellation and the separate file-publication boundary.

A CLI source with no body records never opens its native writer and produces an empty file, including when envelope reconstruction is selected. This differs from explicitly finalizing a library array writer, which emits [] and an LF. An explicitly opened empty envelope document retains its declared framing and has a body count of zero.

XML Format

The XML reader selects record elements by a slash-separated path of element names and maps each one onto the source’s declared schema:. Child elements bind to fields by name; attributes bind under a configurable prefix. Namespaces are stripped by default so schema field names stay clean. See Source Nodes for the shared schema and transport rules.

- type: source
  name: catalog
  config:
    name: catalog
    type: xml
    path: "./data/catalog.xml"
    schema:
      - { name: product_id, type: int }
      - { name: name, type: string }
      - { name: price, type: float }
    options:
      record_path: "catalog/product"    # slash-separated element path
      attribute_prefix: "@"             # prefix for XML attribute fields
      namespace_handling: strip         # strip | qualify
      max_index_bytes: 64MB             # cap on retained envelope sections (optional)

Text encoding

XML input must contain valid UTF-8. One leading UTF-8 BOM is removed on every physical file open, including the envelope pre-scan. UTF-16/32 BOMs are rejected. An XML declaration may omit encoding or declare UTF-8; other encodings and conflicting declarations are rejected. Convert such input to UTF-8 rather than adding an encoding option. Names, attributes, text and CDATA are never decoded with replacement characters.

Validation applies to bytes the reader consumes. A pre-scan may find a late error before any body record is delivered; a streaming body can have already delivered earlier records. Each subsequent file establishes its own BOM and declaration policy. Metadata adjacent to a selected record does not become an extra row, and repeated matching containers preserve body order and empty rows.

Output is UTF-8 without a BOM or XML declaration. See native document boundaries for envelope and empty-output behavior.

Options