Parquet Storage Schema #

Each Litmus run produces one Parquet file. Every row carries an explicit record_type discriminator with one of three values:

• record_type = 'run' — exactly one row per file. Carries run-level identity, timing, outcome, plus DUT / station / project / git / environment context.
• record_type = 'step' — one row per (step_path, vector_index) execution. Step identity, timing, outcome, dynamic in_* / out_* columns. Measurement columns are NULL.
• record_type = 'measurement' — one row per recorded measurement. Carries the measurement payload plus the same denormalized step + run + DUT + station + fixture context as the corresponding step row.

Step and measurement rows share grain (run_id, step_path, vector_index); measurement rows are further keyed by measurement_name. A step that records N measurements emits 1 step row + N measurement rows.

The canonical schema lives at src/litmus/data/schemas.py (RUN_ROW_SCHEMA); this page is a human-readable mirror of it.

File layout #

<data_dir>/runs/{date}/
├── {timestamp}_{serial}.parquet           # Run row + all step + measurement rows for one run
├── {timestamp}.parquet                    # Same shape, no DUT serial (dev runs)
└── {timestamp}_{serial}_ref/              # Reference data (waveforms, images, files)
    ├── {vector_id}_scope_waveform.npz
    ├── {vector_id}_camera_image.png
    └── ...

Timestamps are UTC and sort naturally. DuckDB / Spark / Polars / Pandas all read the file directly with read_parquet.

Discriminator #

Every query starts here. Three values:

• run — one row per run carrying run-level metadata (start/end timestamps, DUT serial, station, outcome).
• step — one row per (step, vector) combination.
• measurement — one row per measurement name within a (step, vector).

To list steps: WHERE record_type = 'step'. To list measurements: WHERE record_type = 'measurement'. All kinds: omit the filter.

Identity & timing #

Who — operator #

What — DUT #

What — product spec #

Where — station #

Where — fixture #

Where — instruments (dynamic step_instruments_*) #

Per-step instrument identity, captured from the pytest fixtures the test actually used. All columns are list[string] (one entry per instrument) and arrays stay in parallel order.

For real hardware, identity comes from *IDN? at session start. For mock instruments, identity comes from the instrument YAML configs.

-- DuckDB: unnest parallel arrays for per-instrument queries
SELECT
    step_name,
    unnest(step_instruments_name) AS instrument,
    unnest(step_instruments_serial) AS serial,
    unnest(step_instruments_cal_due) AS cal_due
FROM read_parquet('data/runs/**/*.parquet')
WHERE record_type = 'step';

Test context #

Input conditions (dynamic in_*) #

For each parametrize axis or sidecar sweep parameter, the writer emits a column. Types are inferred from values.

Naming convention:

Bare names are spec-relevant for condition matching; prefixed names are stimulus/settings. Convention is enforced by docs, not by the writer.

Observations (dynamic out_*) #

Observations are measured context — readings captured during the test, not commanded values.

Examples: out_temp_probe.temperature, out_temp_probe.humidity, out_scope.waveform.

For non-scalar payloads, the value is a file://_ref/... URI:

from litmus.data.backends.parquet import load_file, is_file_reference
 
if is_file_reference(column_value):
    data = load_file(parquet_path, column_value)

Measurement core (on record_type='measurement' rows) #

Limits (on record_type='measurement' rows) #

Spec traceability #

-- Yield by characteristic across all products
SELECT characteristic_id, product_id,
       AVG(CASE WHEN measurement_outcome='passed' THEN 1.0 ELSE 0.0 END) AS yield
FROM read_parquet('data/runs/**/*.parquet')
WHERE record_type = 'measurement'
GROUP BY characteristic_id, product_id;

Measurement signal path #

Rollup outcomes #

Environment traceability #

Custom metadata #

Test code can add arbitrary columns via run_context.set():

def test_example(run_context, psu, dmm, verify):
    run_context.set("operator_badge", "EMP-12345")
    run_context.set("fixture_serial", "FIX-001")
    run_context.set("ambient_temp", 23.5)
    ...

Those become Parquet columns prefixed custom_* with inferred types.

Outcome values #

Source of truth: src/litmus/data/models.py (Outcome).

Comparator values #

Retries #

All retries are stored. Each retry produces measurement rows with the same vector_index and an incremented vector_retry:

vector_index | vector_retry | measurement_name | measurement_value | measurement_outcome
0            | 0            | output_voltage   | 3.50              | failed   ← first execution
0            | 1            | output_voltage   | 3.48              | failed   ← first retry
0            | 2            | output_voltage   | 3.30              | passed   ← second retry

Filter to the final execution with WHERE vector_retry = (SELECT MAX(vector_retry) ...) or use the daemon's runs view, which already rolls retry_count per (run_id, step_path, vector_index).

File-level metadata #

Beyond columns, each Parquet file carries metadata:

import pyarrow.parquet as pq
from litmus.environment import EnvironmentSnapshot
 
pf = pq.ParquetFile("data/runs/2026-05-16/20260516T143025Z_SN001.parquet")
metadata = pf.schema_arrow.metadata
env = EnvironmentSnapshot.model_validate_json(metadata[b"environment_json"])
print(f"Python {env.python_version}, Litmus {env.litmus_version}")

Querying examples #

Load a run with pandas #

import pandas as pd
 
df = pd.read_parquet("data/runs/2026-05-16/20260516T143025Z_SN001.parquet")
 
# Step rows
steps = df[df["record_type"] == "step"]
# Measurement rows with full context
measurements = df[df["record_type"] == "measurement"]
 
# Failures with full context
failures = measurements[measurements["measurement_outcome"] == "failed"]
print(failures[["step_name", "measurement_name", "measurement_value",
                "limit_low", "limit_high", "dut_pin", "instrument_name"]])

Yield by station with DuckDB #

SELECT
    product_id,
    station_id,
    measurement_name,
    COUNT(*) AS total,
    SUM(CASE WHEN measurement_outcome = 'passed' THEN 1 ELSE 0 END) AS passed,
    ROUND(100.0 * SUM(CASE WHEN measurement_outcome = 'passed' THEN 1 ELSE 0 END) / COUNT(*), 2) AS yield_pct
FROM read_parquet('data/runs/**/*.parquet')
WHERE record_type = 'measurement'
GROUP BY 1, 2, 3
ORDER BY yield_pct ASC;

Cross-run instrument-failure correlation #

SELECT
    instrument_name,
    instrument_resource,
    COUNT(*) AS failures
FROM read_parquet('data/runs/**/*.parquet')
WHERE record_type = 'measurement'
  AND measurement_outcome = 'failed'
GROUP BY 1, 2
ORDER BY failures DESC;

Slowest steps across runs #

SELECT
    step_name,
    AVG(EPOCH(step_ended_at) - EPOCH(step_started_at)) AS avg_seconds,
    COUNT(*) AS runs
FROM read_parquet('data/runs/**/*.parquet')
WHERE record_type = 'step'
  AND step_started_at IS NOT NULL
GROUP BY step_name
ORDER BY avg_seconds DESC;

ATML / IEEE 1671 alignment #

See also #

• Models — Pydantic model index + ERD
• Event types — the event-log payloads that source these rows
• Measurement traceability — how the signal-path columns get populated