datasets

Datasets are the output tables. Each one selects messages, decodes their payloads, derives identity and time, and declares the columns every row will have. Declared schemas are required — this is what makes Parquet output deterministic and schema evolution explicit.

datasets:
  - id: telemetry
    match: { source: fleet-mqtt, topic: "thadam-demo/v1/+/+/telemetry" }
    decode: json
    identity:
      site:  { id: 100, template: "{topic.2}" }
      robot: { id: 101, template: "{topic.3}" }
    event_id: { key: [site, robot, seq] }
    dedup: { window: 10m, max_keys: 500k }
    explode: { path: /samples, index_field: sample_idx }
    time: { field: /ts_us, unit: us, domain: utc, max_skew: 5s }
    fields:
      - { id: 1000, name: seq,        type: u64,        from: /seq, required: true }
      - { id: 1002, name: voltage_v,  type: f32,        from: /voltage_mv, unit: mV, convert_to: V }
      - { id: 1005, name: temps_c,    type: "list<f32>", from: /temps_dc,   unit: dCel, convert_to: Cel }

Dataset keys

match

identity

Each level is name: { id, template }:

• Level names start with a letter and use letters, digits, _ (a leading _ is reserved; envelope columns start with _).
• At most 32 levels per dataset (MAX_ASSET_LEVELS).
• Levels become columns in every row of the dataset.
• Because identity is a hierarchy, it must keep the written order all the way to the transform chain; it is hashed and passed as an ordered list, never an unordered map.

event_id

event_id: { key: [site, robot, seq] }

Without event_id, IDs are derived from the WAL position (xxh3_128 over epoch, frame sequence, record index): deterministic across replay, but different for a broker redelivery. With a content-derived key, a redelivered message hashes to the same ID as the original — which is exactly what deduplication needs.

dedup

dedup: { window: 10m, max_keys: 500k }

• Runs at message level, before explode, so one large message costs one key, not one key per element.
• Warm-started from the WAL tail [cursor − window, cursor) at startup, so dedup survives restarts deterministically.
• Best-effort by design (ADR-0011): a duplicate arriving after the window, or after max_keys eviction, is not detected. Authoritative deduplication belongs downstream (ReplacingMergeTree, QUALIFY, or DISTINCT ON).
• dedup requires event_id.key. Validation fails with config.invalid without it: the default WAL-position ID changes when a message is redelivered, so duplicates would never be recognized (ADR-0011).
• Memory shows up in the thadam validate estimate as max_keys × 48 B (upper bound).

explode

explode: { path: /samples, index_field: sample_idx }

Inside an exploded dataset, a field’s from may use ^/ to read from the parent message (for example a sequence number that applies to every sample): from: "^/seq". ^/ is rejected in datasets without explode.

time

time: { field: /ts_us, unit: us, domain: utc, max_skew: 5s }

Exactly one of field (read the payload) or attr (read a message attribute) is required; giving both or neither is a config error.

time: { field: /ts, unit: rfc3339, default_offset: "+02:00", domain: utc }

Without default_offset, RFC 3339 text that carries no offset is a runtime reject (the message is quarantined).

Examples for the attribute and ROS 2 forms:

# ROS 2 builtin_interfaces/Time (header.stamp), already unix nanoseconds
time: { field: /header.stamp, unit: ns, domain: utc, max_skew: 5s }

# MCAP publisher clock, attached by the source
time: { attr: mcap.publish_time, unit: ns, domain: utc, max_skew: 5s }

# MCAP recorder clock
time: { attr: mcap.log_time, unit: ns, domain: utc }

mcap.log_time is recorder time: when the recorder wrote the message. mcap.publish_time and header.stamp are publisher time: the clock of the node that produced the message, which can drift from the recorder or the host. Neither is corrected; max_skew flags a gap against ingest time.

For attr, a missing attribute is the same as an absent payload field: the event is flagged MISSING_TIME. Attribute text that does not parse in unit is a reject (decode.invalid_time), the same as an unparseable payload timestamp.

fields

Each entry declares one column:

Parquet field IDs

Every column carries a stable Parquet field_id from the first release, allocated from disjoint ranges (ADR-0013):

Rules: IDs are unique per dataset, ranges are enforced per kind, and an ID is never reused for a different column — even after the column is removed. Thadam cannot see history; retiring IDs is the configuration owner’s job. This is what keeps the Iceberg fast_append path cheap (no rewrite) and makes schema evolution by field ID (add optional columns, widen int→long, float→double, or rename a known ID; never narrow or reuse an ID) safe.

Complete example

The six-dataset demo with dedup, explode, units, and quality transforms is examples/edge-capture/thadam/pipeline.yaml.