transforms and predicates

The transforms: list applies quality, projection, and filtering steps to datasets, in listed order, per sink chain. Every step takes a dataset selector and an optional sinks restriction.

Config version 1 accepts all eleven kinds the transform engine implements: validate, on_change, filter, sample, enrich, rename, project, cast, pose, joint_state, and window (experimental). See also the transform reference.

Each step is checked against the dataset’s fields as left by the earlier steps for that dataset (and for the sinks the step applies to). After rename, project, cast, or enrich, later steps use the new names and types.

Step shell

transforms:
  - kind: validate | on_change | filter | sample | enrich | rename | project | cast | pose | joint_state | window
    dataset: telemetry            # the dataset the step applies to (required, single id)
    sinks: [local]                # optional: restrict the step to these sink ids
    ...                           # kind-specific keys

kind: validate

Assigns quality flags and a status to events whose fields break a range rule. Events are flagged, not dropped — quality is data.

- kind: validate
  dataset: telemetry
  rules:
    - { field: motor_temp_c, min: -40, max: 150, flag: OUT_OF_RANGE, status: bad }
    - { field: temps_c, min: -40, max: 150, flag: OUT_OF_RANGE, status: uncertain }

Each rule:

A null field is not checked. Status and flag land in the _quality_status, _quality_flags, and _quality_reason columns.

kind: on_change

Passes an event only when a watched field changed value, or when the heartbeat elapses — deadband-style reduction for slow-changing state. It is lossy by design and it is not deduplication (product.md UC-05 conflates the two; ADR-0011 keeps them separate).

- kind: on_change
  dataset: state
  fields: [mode, cycle_id, program]
  heartbeat: 5m

An entry is either a bare name or a mapping:

fields:
  - mode
  - { field: soc_pct, deadband: 0.5 }

deadband: 0 is the same as the bare name and hashes identically.

kind: filter

Drops events that do not match a structured predicate; dropped events are counted as filtered, and they are recoverable by replaying with a corrected predicate.

- kind: filter
  dataset: events
  where: { field: level, ne: debug }

Predicate grammar

Structured YAML predicates (ADR-0018) — composable, type-checked at config time, deterministic in replay.

# One condition
where: { field: level, ne: debug }

# Combinators
where:
  all:
    - { field: level, ne: debug }
    - any:
        - { field: site, eq: plant-2 }
        - { field: soc_pct, min: 10 }
  # or:  not: { field: code, in: [E100, E200] }

Combinators: all (every child true), any (any child true), not (child false). A condition names a field and exactly one test:

Rules (checked by thadam validate):

• Fields resolve against the dataset: declared fields, identity levels, the explode index field, and the quality names quality.status, quality.flags, quality.reason. Typo’d names are errors with suggestions.
• Literal typing: integer fields take integer literals that fit (u64 values above i64::MAX stay exact, never widened to float); float fields compare in their own domain.
• Null semantics: a null or missing field makes eq, the ordered comparisons, in, min, and max false, and ne true. NaN is unordered. To test for null, use the null: test; eq: null is rejected.
• null test: allowed on any resolvable name — fields, identity levels, the explode index field, and quality.*. A condition still takes exactly one test. null: needs no quoting in YAML.
• Cost limits: predicate nodes, depth, and in-set size are bounded by chain limits; overflow is a configuration error, not a runtime surprise.
• An evaluation that still cannot complete at runtime becomes a transform.eval_failed reject with the raw bytes kept.

kind: sample

Thins events per asset, in ingest order. Lossy by design; dropped events are not recoverable except by replay.

- kind: sample
  dataset: imu
  every_n: 10
  min_interval: 100ms

At least one of every_n and min_interval is required (config.invalid otherwise). With both, an event must pass the count rule first, then the interval rule.

kind: enrich

Adds fields from a static CSV or YAML lookup table.

- kind: enrich
  dataset: telemetry
  file: robots.yaml
  key: { field: robot, column: robot_id }
  add:
    - { name: payload_kg, type: f32, unit: kg }
    - { name: cell, type: str, column: cell_name }
  on_miss: null

key:

Each add entry:

Table layout:

• CSV has a header row. CSV cannot hold list<...> columns; use YAML for those.
• YAML is either a list of mappings (each holding the key column) or a mapping from key to a mapping of columns.

The file is read when the transform chain is compiled at startup, not when thadam validate checks the config, so a missing or malformed file is not a config-validation error. Its SHA-256 is recorded for lineage, as the enrichment hash that replay row-set equivalence depends on. Added fields have no Parquet field ID declared in config.

kind: rename

Renames fields. Values are untouched.

- kind: rename
  dataset: telemetry
  fields: { temp_c: temperature_c, volts: voltage_v }

Renames apply simultaneously, so swapping two names is allowed. Renaming a field to its own name is a no-op.

kind: project

Keeps only the listed fields.

- kind: project
  dataset: telemetry
  fields: [seq, soc_pct, voltage_v]

kind: cast

Widens field types. Narrowing is an error.

- kind: cast
  dataset: telemetry
  fields: { seq: i64, voltage_v: f64 }

Allowed casts:

• Signed integers widen: i8 → i16 → i32 → i64. Unsigned integers widen likewise: u8 → u16 → u32 → u64.
• Any unsigned integer to a larger signed integer (for example u32 → i64).
• f32 → f64, list<f32> → list<f64>, and list<i32> → list<i64>.
• A cast to the field’s current type is allowed and does nothing.

Anything else is a config.invalid error: narrowing (f64 → f32, i64 → i32), integer to float, and anything involving str, bool, ts, or bytes. The engine reports the same condition at chain-compile time as schema.narrowing.

kind: pose

Flags malformed flattened poses — the dotted fields the ROS 2 decoder writes for geometry_msgs/PoseStamped and cousins (see robotics conventions). Events are flagged, not dropped — quality is data.

- kind: pose
  dataset: poses
  # prefix: pose                 # the default
  # frame_field: header.frame_id # the default; an explicit null skips the frame check
  # quaternion_tolerance: 0.001  # the default

With prefix: pose the step reads pose.position.{x,y,z} and pose.orientation.{x,y,z,w}, each an f32 or f64 field.

The norm is only computed when all four quaternion components are finite numbers. An event violating several rules raises every flag; the reason is the first violation (an upstream reason is kept), and everything lands in _quality_status, _quality_flags, and _quality_reason as usual.

The flattened fields are declared as ordinary (dotted) dataset fields, so thadam validate checks them here: every <prefix>.position.{x,y,z} and <prefix>.orientation.{x,y,z,w} must exist as an f32 or f64 field, and the frame field — header.frame_id by default — as a str field. Each missing or mistyped one is its own error at transforms[…].prefix or transforms[…].frame_field, and the scalar options are checked too (prefix non-empty, the tolerance range, a non-empty frame_field). The engine re-checks against the actual record shape when the transform chain compiles at pipeline startup.

kind: joint_state

Flags malformed joint-state arrays, following sensor_msgs/JointState semantics (see robotics conventions). Events are flagged, not dropped — quality is data.

- kind: joint_state
  dataset: joints
  # name_field: name         # the default
  # position_field: position # the default
  # velocity_field: velocity # the default
  # effort_field: effort     # the default

Array fields accept list<f32> and list<f64> (generic lists of floats decode as either); a null array reads as an empty one. thadam validate resolves all four fields against the dataset as left by earlier steps — unknown names and wrong element types are configuration errors — and the engine re-checks against the actual record shape when the chain compiles at startup.

Several violations at once settle on the first reason in this order; flags accumulate.

kind: window

Status: Available (experimental). Tumbling or sliding windows on ingest time (time: ingest, the default) or event time (time: event, with per-asset watermarks and allowed_lateness), one aggregate event per asset and window. The design is in docs/design/advanced-processing.md.

datasets:
  - id: telemetry
    # ...
    fields:
      - { id: 1000, name: motor_temp_c, type: f32 }
      # The aggregates the window writes. The Parquet sink writes the columns
      # a dataset declares, so declare one per output column (see below).
      - { id: 1010, name: temp_mean, type: f64 }
      - { id: 1011, name: temp_max, type: f64 }
      - { id: 1012, name: samples, type: u64 }
      - { id: 1013, name: window_start, type: ts, unit: ns }
transforms:
  - kind: window
    dataset: telemetry
    size: 1m
    slide: 15s                 # optional: omit or set equal to size for tumbling windows
    time: event                # or ingest (the default)
    allowed_lateness: 5s       # event time only (0s when absent)
    aggregates:
      - { field: motor_temp_c, fn: mean, as: temp_mean }
      - { field: motor_temp_c, fn: max, as: temp_max }
      - { field: motor_temp_c, fn: count, as: samples }

Each aggregate:

What a window emits. Events accumulate into panes of width slide (size when slide is omitted), and each non-empty window [w·slide, w·slide + size) merges its size / slide panes (stddev uses Chan-Welford parallel merge so the merged result equals population stddev over all samples in the window). With time: ingest, an asset’s window closes when the asset sends an event whose ingest time falls at or past window_end. With time: event, each asset tracks its own watermark (max_event_time − allowed_lateness); an out-of-order event within allowed_lateness joins its pane, and open windows with window_end ≤ watermark close and emit in ascending window_start order (a pane is retained until every sliding window covering it has closed). The emitted event has window_start and window_end (ts), then one column per aggregate: count is a u64, the rest are f64 (null when the window saw no value). A null or NaN value is skipped. The event replaces the dataset’s fields, so later steps see only these columns, and the asset identity stays. Its event time is the window start, its ingest time is the closing event’s, and its WAL position is the closing event’s frame. Its id is derived from the dataset, asset, window start, size, and column names, so a re-read after a restart gives the same id. The events that fed the window are not emitted.

Counters and quality. Events absorbed into an open window, and events with no UTC event time under time: event (for example a monotonic or sim time domain), count as filtered. Under time: event, an event whose window has already closed (window_end ≤ watermark) is dropped without re-emitting and counted as late (thadam_chain_events_total{outcome="late"}). A window that was open across a WAL gap emits with quality.status = uncertain and quality.reason = "window_gap"; when thadam replay starts after frame 1 (--from > 1), the first window opened for each asset emits with quality.status = uncertain and quality.reason = "window_range_start".

Columns reach Parquet only when declared. The Parquet sink writes the columns the dataset declares, matched by name. A window’s output columns are written only if the dataset declares fields with the same names (as in the example above, where the declared fields only exist to name and type the output columns; they read null from the payload). The same holds for columns added by enrich.

Open windows and the cursor.

• There are no wall-clock timers. Without idle_close, an asset’s open windows stay open until that asset sends an event that advances past them; nothing is flushed on shutdown. With idle_close, another asset’s event whose ingest time is at least idle_close past the silent asset’s last event closes all of that asset’s open windows (ordered by (first_pos, asset) across assets and ascending window_start within an asset), advances its watermark to the last closed window_end, and releases its hold on the cursor.
• Open windows are kept in memory only. To survive a restart without a new state file, the sink’s WAL cursor is held back: it never moves to or past the first frame of a window whose aggregate the sink does not hold yet, and a restart re-reads from the cursor to rebuild the windows. Aggregates the sink already holds are skipped, so a restart neither drops an aggregate nor writes one twice.
• The WAL therefore keeps data for at least the window length (plus allowed_lateness under time: event), plus the time any asset stays silent (bounded by idle_close when other assets in the dataset keep speaking). Without idle_close, a silent asset’s open window holds the cursor for every dataset in that sink’s chain until the asset speaks again, so the WAL grows (or, with on_full: block, blocks) in the meantime. State is bounded by the chain’s per-step state limit (max_state_keys, 100,000 assets by default); beyond it, events of further assets are rejected to quarantine with a transform.failed reason rather than dropped.
• Under time: ingest, if ingest time steps backwards, the event joins its asset’s open window.
• Restarts are exact for the window step itself. A step before it that depends on history (sample every_n, on_change, dedup) is rebuilt from warm_start_frames of history, so a window behind such a step can differ from an uninterrupted run when the history is longer than that. Steps after a window start empty after a restart.

Limits in this release.

• Only segment sinks (such as parquet) support window. Building a pipeline that routes a windowed dataset to a batch sink is a configuration error.
• Late events under time: event are dropped and counted rather than emitting retractions or corrections.
• Tumbling and fixed-slide windows only (size / slide ≤ 1024); session windows, median, and percentiles are not implemented.

thadam validate checks size, slide, time, allowed_lateness, idle_close, the aggregates, and the field types, includes the per-step open-window state bound (scaled by size / slide when slide is set) and cursor hold-back span (size + allowed_lateness) in its memory estimate, and checks the later steps against the window’s columns.

Anomaly rules on windowed signals

Because a window step replaces the dataset’s fields with its aggregate columns and passes each emitted aggregate event through the steps after it, anomaly rules over a time window need no dedicated operator — compose window with filter (to emit only anomalous windows) or validate (to keep every window and flag violations in _quality_status and _quality_flags):

transforms:
  - kind: window
    dataset: motor_anomalies
    size: 1m
    time: event
    allowed_lateness: 5s
    aggregates:
      - { field: motor_temp_c, fn: mean, as: temp_mean }
      - { field: motor_temp_c, fn: range, as: temp_range }
      - { field: motor_temp_c, fn: stddev, as: temp_stddev }
      - { field: motor_temp_c, fn: delta, as: temp_delta }
  - kind: validate
    dataset: motor_anomalies
    rules:
      # Sustained high temperature over 1 minute
      - { field: temp_mean, max: 85.0, flag: OUT_OF_RANGE, status: bad }
      # Flat-lined sensor (range == 0 over a full minute) or excessive noise
      - { field: temp_range, min: 0.05, flag: STALE, status: uncertain }
      - { field: temp_stddev, max: 12.0, flag: OUT_OF_RANGE, status: uncertain }
      # Rapid rise or fall across the window (last - first)
      - { field: temp_delta, min: -15.0, max: 15.0, flag: OUT_OF_RANGE, status: uncertain }

Feature extraction for ML pipelines

A window step over high-rate sensor telemetry produces fixed-cadence feature vectors (count, min, max, mean, stddev, range, delta, rms) as a declared dataset for downstream training or scoring:

transforms:
  - kind: window
    dataset: imu_features
    size: 1s
    time: event
    allowed_lateness: 200ms
    aggregates:
      - { field: accel_z, fn: count, as: accel_z_n }
      - { field: accel_z, fn: mean, as: accel_z_mean }
      - { field: accel_z, fn: stddev, as: accel_z_stddev }
      - { field: accel_z, fn: range, as: accel_z_range }
      - { field: accel_z, fn: delta, as: accel_z_delta }
      - { field: accel_z, fn: rms, as: accel_z_rms }

Spectral features (such as FFT) and in-chain model inference are Planned (P2; see docs/design/advanced-processing.md).

Execution position

Transforms run after decode, event-ID derivation, dedup, explode, field extraction, unit conversion, and time handling — see per-sink chains for the full order and for how panics and poison records are handled.