Transform reference

Transforms are the synchronous, pure steps a per-sink chain runs between decode and delivery. This page covers every step kind implemented in thadam-transform, with its options and examples. All eleven kinds are accepted in the top-level transforms: list of config version 1.

See also the configuration reference, which has the same information organized by config key.

Execution model recap

• Steps run per chain, per dataset, after decode → event-ID/dedup → explode → field extraction/unit conversion → time handling, and each step’s dataset: (and optional sinks:) selector is applied.
• Steps are deterministic: no wall clock, no randomness. Sampling uses arrival order and ingest time.
• Cost is bounded by ChainLimits (shape registry, dedup keys, JSON depth, predicate nodes/depth/set size).

validate — range rules to quality flags

Config: accepted. Assigns a quality flag and status when a numeric field (or every element of a numeric list) falls outside min/max; events are flagged, not dropped. Full key table: kind: validate.

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

on_change — deadband with heartbeat

Config: accepted. Keeps an event only when a watched field changed, or when heartbeat of ingest time passed. Lossy by design; not deduplication (ADR-0011). Full key table: kind: on_change.

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

Watched fields may be bare names or { field: <name>, deadband: <number> } entries with an absolute numeric deadband (default 0); both forms are accepted in config.

filter — structured predicates

Config: accepted. Drops events failing a structured predicate (ADR-0018). Full grammar: predicates.

- kind: filter
  dataset: events
  where:
    all:
      - { field: level, ne: debug }
      - any:
          - { field: site, eq: plant-2 }
          - { field: soc_pct, min: 10 }

Tests: eq, ne, lt, le, gt, ge, in, min, max, and null (null: true tests for null-or-missing); config version 1 accepts all of them. Combinators: all, any, not. expr: is reserved and refused. Names resolve against dataset fields, identity levels, the explode index field, quality.status, quality.flags, and quality.reason; missing/null fields make ordered tests false and ne true.

sample — every-N and minimum-interval

Config: accepted. Keeps the first event and every Nth after it, and/or events at least an interval apart (ingest time), per asset. Full key table: kind: sample.

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

enrich — static table lookup

Config: accepted. Adds fields from a static CSV or YAML table, read once at chain-compile time; the file’s SHA-256 is recorded as the enrichment hash in lineage (one of the three inputs of replay row-set equivalence). Relative paths resolve against the config file’s directory. Full key table: kind: enrich.

- 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        # null (default) | reject | drop

rename — field rename

Config: accepted. Renames fields without touching values. Full key table: kind: rename.

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

project — keep a subset

Config: accepted. Keeps only the listed fields, in the listed order. Full key table: kind: project.

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

cast — widening only

Config: accepted. Widens field types (i32 → i64, f32 → f64, and so on). Narrowing is rejected: config.invalid at validation and a compile error (schema.narrowing) in the engine, matching the schema-evolution rules for datasets. Full key table: kind: cast.

- kind: cast
  dataset: telemetry
  fields: { soc_pct: f64 }

pose — flattened-pose checks

Config: accepted. Flags a flattened pose (default prefix pose, the ROS geometry_msgs/PoseStamped layout) whose components are missing (pose_missing_field) or non-finite (pose_not_finite), whose quaternion norm drifts past quaternion_tolerance (quaternion_not_unit, OUT_OF_RANGE), or whose frame id (frame_field, default header.frame_id, explicit null skips) is empty (missing_frame_id). Flagged, not dropped; see robotics conventions. Full key table: kind: pose.

- kind: pose
  dataset: poses
  quaternion_tolerance: 0.01

joint_state — joint-array checks

Config: accepted. Flags a joint-state record (ROS sensor_msgs/JointState semantics) whose position/velocity/effort arrays are neither empty nor as long as name (joint_state_length_mismatch), whose names repeat (duplicate_joint_name), or with non-finite values (joint_state_not_finite). Field names are configurable (name_field, position_field, velocity_field, effort_field). Full key table: kind: joint_state.

- kind: joint_state
  dataset: joints

window — tumbling and sliding windows (experimental)

Config: accepted. Status: Available (experimental). Aggregates (count, sum, min, max, mean, first, last, stddev, range, delta, rms) per asset over tumbling or sliding windows (size, optional slide) of ingest time (time: ingest, default) or event time (time: event, with per-asset watermarks and allowed_lateness), emitting one event per asset and window when the window closes (or on idle_close). It is the only step that keeps state across events that a restart cannot rebuild by warm-up, so the sink’s WAL cursor is held back behind open windows and a restart re-reads from there. Segment sinks only. Full key table, limits, and the retention cost: kind: window.

- kind: window
  dataset: telemetry
  size: 1m
  aggregates:
    - { field: motor_temp_c, fn: mean, as: temp_mean }
    - { field: motor_temp_c, fn: max, as: temp_max }

Fixed-position behaviors (not transforms: steps)

These dataset-shaping behaviors are configured on the dataset, not in transforms:, and run in a fixed order before your steps: