Add Parquet storage schema and a PyArrow example

README.md

@@ -51,6 +51,9 @@ implement a missing feature.
 
 - `docs/design/benchmark-result-schema.md` — the benchmark result schema and
   migration contract (the core design document).
+- `docs/design/measurement-result-parquet.md` — how a result is stored in
+  Parquet (`src/benchx/parquet.py`), with a small converter in
+  `tools/json_to_parquet.py`.
 - `docs/user_stories/` — user story template and collected stories.
 - `src/benchx/` -- an in-progress implementation, with its tests in `tests/`.
 - `src/benchx/schemas/` -- JSON schemas for the benchmark result and other components

docs/benchx.md

@@ -233,7 +233,8 @@ every result is pyperf. Results also record the installed project's version
   (`paths.store` in `benchx.conf.json` moves it). It reads documents strictly, validates them, and applies the
   schema §5.4 contract: idempotency and the rejection codes `malformed`,
   `idempotency-conflict`, `unit-conflict`, `identity-violation`, and
-  `attempt-conflict`. Each result is one row: the columns #30 derives from the
+  `attempt-conflict`. Each result is one row: the columns
+  [`measurement-result-parquet.md`](design/measurement-result-parquet.md) derives from the
   schema, the store's derived columns, the canonical document, and the work
   order it references. Entities and series are computed from the rows on
   query, under the identity policy in `benchx/identity.py`: the default of

docs/design/README.md

@@ -14,9 +14,9 @@ designed around it (see the repository README for the philosophy).
 | [`comparator.md`](comparator.md) | Verdicts in history mode and run mode: the eligibility guard, dispersion-scaled verdicts, the comparison document | draft |
 | [`prototype-scope.md`](prototype-scope.md) | What the minimal implementation demonstrates and the decisions locked for it | agreed |
 | [`prototype-design.md`](prototype-design.md) | How the implementation is built: one CLI over a Python API, a Parquet store, three JSON documents | draft |
+| [`measurement-result-parquet.md`](measurement-result-parquet.md) | The Parquet form of a result: the mapping from the result schema to Arrow types, the message columns of the store's row | draft |
 
-In review as pull requests: the work order (`work-order.md`) and the
-Parquet form of a result (`measurement-result-parquet.md`, #30).
+In review as a pull request: the work order (`work-order.md`).
 
 Not yet designed (roles fixed in the decomposition, documents pending):
 workbench, scheduler internals, change detector, alerting & annotations,

docs/design/measurement-result-parquet.md

@@ -0,0 +1,173 @@
+# Measurement Results in Parquet
+
+**Status:** Draft for review
+
+**Companion to:** `benchmark-result-schema.md` §5 (the message; where this document and the schema differ, the schema wins), `prototype-design.md` §4 (the local store, whose row begins with these columns)
+
+**Author:** Rok Mihevc
+
+## 1. Scope
+
+This document defines the columns under which a validated measurement result (§5.2 ingest object) is stored in Parquet: the mapping from the message JSON Schema to Arrow types, implemented in `src/benchx/parquet.py`, and a small program that writes them from JSON files, `tools/json_to_parquet.py`. It defines nothing about the message itself.
+
+One row is one result, that is one attempt and one quantity. These *message columns* hold the reported facts only. Fingerprints, derived estimates, and entity tables are computed by a store and sit beside them (§4). The mapping is not byte-preserving, so a store that needs the original document keeps it beside the columns, as the local store does in its `document` column.
+
+## 2. Mapping
+
+The Arrow schema is derived from `src/benchx/schemas/measurement-result/0.1.0/schema.json` by the rules below, so it follows the message schema without being maintained by hand.
+
+| JSON Schema node | Arrow and Parquet type |
+|---|---|
+| closed object (`additionalProperties: false`) | `struct` with the same field order; required properties are non-nullable |
+| open object, or a value of mixed primitive types | `json`: the Arrow JSON extension type, written with the Parquet JSON logical type, holding RFC 8785 canonical text |
+| `oneOf` of closed objects (`summaries`) | one `struct` with every property any alternative declares, in first-appearance order; non-nullable only when every alternative requires it; alternatives that give one property different types are an error, not a silent choice |
+| `oneOf` of number and structured observation (`observations`) | the structured observation's `struct`; a bare number is stored with only `value` set |
+| `allOf` of a reference and extra assertions (`derivation`) | the referenced type |
+| array | `list` |
+| string with `format: date-time` | `timestamp[us, UTC]` |
+| string, enum, string const | `string` |
+| integer, integer const | `int64` |
+| number | `float64` |
+| boolean | `bool` |
+| any other `const`, `enum`, or `oneOf` | an error, so that a schema change the rules do not cover is noticed rather than stored loosely |
+
+Open objects stay JSON because their keys are project-defined: workload parameters, subject configuration, the protocol, environment identity and system description, resource selection, labels, and procedure. Canonical text makes equal values equal strings. A store promotes a key to a typed column only when a query needs it.
+
+## 3. Resulting schema
+
+For message schema 0.1.0, whose messages carry `schema_version: "benchx/measurement-result/0.1.0"`. `*` marks a non-nullable field.
+
+```text
+schema_version*: string
+producer*: struct
+  name*: string, version*: string, mapping_version*: string
+ingest_key*: string, attempt_key*: string, project: string
+source*: struct
+  uri*: string, type: string
+revision*: struct
+  key*: string, native_order: int64, native_time: timestamp[us, UTC]
+benchmark: struct
+  source*: struct                       (as source)
+  revision: struct                      (as revision)
+coordinates*: struct
+  workload*: struct
+    name*: string, parameters: json
+    dataset: struct
+      name*: string, version: string, sha256: string, parameters: json
+  subject*: struct
+    name*: string
+    components*: list of struct
+      role*: string, source: string, revision: string
+    configuration: json
+  protocol*: json
+  environment*: struct
+    schema_version*: string, identity*: json, system: json
+  resource_selection: json
+measurement*: struct
+  quantity*: struct
+    name*: string, unit*: string, direction: string, deterministic: bool
+  kind: string, status*: string, reason: string
+  constraint: struct
+    kind*: string, lower: float64, upper: float64,
+    lower_inclusive: bool, upper_inclusive: bool, cause*: string
+  observations: list of struct
+    value*: float64, ordinal: int64, slot: int64, time: timestamp[us, UTC],
+    group: string, pair: string, included: bool, exclusion_reason: string
+  derivation: struct                    (as estimator)
+  summaries: list of struct
+    type*: string
+    estimator: struct
+      name*: string, method_version*: string, input_level*: string,
+      parameters: json
+      inputs: list of struct
+        role*: string, quantity*: string
+    value: float64, source*: string, name: string, method: string,
+    statistic: string, level: float64, lower: float64, upper: float64,
+    label: string, coverage_factor: float64
+procedure: json
+provenance*: struct
+  run_key: string, labels: json, native_id: string,
+  started_at*: timestamp[us, UTC], ended_at: timestamp[us, UTC],
+  subject_dirty*: string, subject_tree: string,
+  benchmark_dirty*: string, benchmark_tree: string
+  runner: struct
+    name*: string, version: string
+  original_unit: string, source_payload_uri: string,
+  source_payload_sha256: string, ci_uri: string
+  references: list of struct
+    role*: string
+    result*: struct
+      producer*: string, ingest_key*: string
+  artifacts: list of struct
+    kind*: string, media_type*: string, schema: string, uri*: string,
+    sha256*: string
+  info: json
+quality: struct
+  warnings: list of string, validation: json
+```
+
+The file's key-value metadata records which message schema it was written under:
+
+```text
+benchx.schema_uri            = urn:benchx:schema:measurement-result:0.1.0
+benchx.message_schema_sha256 = <SHA-256 of the RFC 8785 form of schema.json>
+```
+
+The checksum matters while 0.1.0 is a draft that changes under one URI.
+
+## 4. Beside the message columns
+
+A store adds the columns it derives at ingest after the message columns, and may copy a key out of an open object into a typed column when a query needs it. The local store adds its project, `reported_coordinates_fingerprint`, `payload_sha256`, an attempt signature, the canonical `document`, and the work order the result references (`parquet.DERIVED`; `prototype-design.md` §4). Those columns and the store's layout belong to the store's design, not to this document.
+
+Because the Arrow schema is derived, a change to `schema.json` changes the file schema, and files written under different message schemas are told apart by `benchx.message_schema_sha256`. An added optional field reads as null from older rows under PyArrow's permissive schema unification, which is how the local store carries old rows over when it rewrites its file; removing a field or changing its type is a new message contract and is not mixed into one file.
+
+## 5. Writing and reading
+
+The program reads each document strictly and checks it as `bx validate` does, then writes all documents given to it as the rows of one file:
+
+```bash
+pip install benchx
+python tools/json_to_parquet.py results.parquet \
+    src/benchx/schemas/measurement-result/0.1.0/examples/*.json
+```
+
+Input is strict because JSON Schema validation only sees a message after it has been parsed, when some defects are already invisible. Messages use the I-JSON data model (RFC 7493), and `benchx.core.loader` rejects what falls outside it before validating:
+
+- **Duplicate object keys.** A parser keeps one of them silently, so two readers could store different values for the same message.
+- **`NaN`, `Infinity`, and `-Infinity`.** Python's parser accepts these tokens although JSON does not, and they have no place in a finite observation.
+- **Integers beyond ±(2^53 − 1).** A larger value is read differently by parsers that hold numbers as binary64, so it would not mean the same thing to every reader, and RFC 8785 cannot serialize it.
+
+RFC 8785 is the canonical form the `json` columns hold and the schema checksum is computed over. One rejected document fails the run and nothing is written.
+
+Any Arrow-capable engine reads the result directly, and the same query runs against the local store, `store.parquet`, since its row begins with these columns. In DuckDB, structs are addressed with dots, the observation list with list functions, and JSON columns with JSON paths:
+
+```sql
+SELECT coordinates.workload.name     AS workload,
+       measurement.quantity.name     AS quantity,
+       len(measurement.observations) AS n,
+       list_min([o.value FOR o IN measurement.observations]) AS minimum,
+       coordinates.protocol->>'$.timer' AS timer
+FROM 'results.parquet'
+ORDER BY workload, quantity;
+```
+
+```text
+workload      quantity         n     minimum   timer
+matmul        cpu-time         10    0.000212  perf_counter
+matmul        gpu-time         10    0.01836   cuda-events
+parquet-read  cpu-temperature  4     61.0      NULL
+parquet-read  wall-time        5     0.0363    perf_counter
+truncate      overhead-ratio   NULL  NULL      perf_counter
+truncate      wall-time        5     0.000128  perf_counter
+```
+
+The temperature is a condition of the wall-time attempt, read from a sensor rather than timed. The overhead ratio is an aggregate-only compound result, so it has no observations.
+
+`tests/test_parquet.py` checks the mapping over the examples (row order, nullability, canonical JSON, observation values, file metadata), the rules that must fail, and the program.
+
+## 6. Not covered
+
+- The rules the message schema leaves to the ingest server or comparator: idempotency and conflicts on `(producer, ingest_key)`, unit immutability, attempt integrity, comparison profiles. The local store implements them (`prototype-design.md` §4).
+- Entity tables, partitioning, and compaction.
+- Reading a message back out of the message columns. Member order, number spelling, and whether an observation was a bare number are not kept; a store that needs the document keeps it (§1).
+- Timestamps finer than microseconds are truncated to the column's precision.

docs/design/prototype-design.md

@@ -1,7 +1,7 @@
 # Prototype Design
 
 **Status:** Draft for review
-**Companion to:** `prototype-scope.md` (what's in the slice); `benchmark-result-schema.md` (the result); `runner-schema.md` and `schemas/work-order/0.1.0` (the work order); `harness-adapter.md`; `benchmark-environments.md`; `comparator.md`; `runner.md`; `system-decomposition.md` §3.4 (the workbench); `measurement-result-parquet.md` (#30, the result row)
+**Companion to:** `prototype-scope.md` (what's in the slice); `benchmark-result-schema.md` (the result); `runner-schema.md` and `schemas/work-order/0.1.0` (the work order); `harness-adapter.md`; `benchmark-environments.md`; `comparator.md`; `runner.md`; `system-decomposition.md` §3.4 (the workbench); `measurement-result-parquet.md` (the result row)
 
 ## 1. Shape
 
@@ -404,15 +404,15 @@ through. There are no side files:
 no JSON, no separate entity tables. Each row is one result, the
 `benchmark_result` of schema §4.1:
 
-- **Message columns:** the row `measurement-result-parquet.md` (#30) derives
+- **Message columns:** the row `measurement-result-parquet.md` derives
   from the result schema, typed structs and lists with open objects as
   canonical JSON, for queries.
 - **Derived columns:** the store project (the result's own, or
   `local/<hostname>` for a thin one), `reported_coordinates_fingerprint`,
   `payload_sha256`, and an attempt signature over the fields schema §5.3
   requires results of one attempt to share.
 - **The document:** `document` holds the result in its RFC 8785 canonical
-  form. #30's columns cannot reproduce it, since its mapping is not
+  form. The message columns cannot reproduce it, since its mapping is not
   byte-preserving (a bare observation and a structured one map alike), so this
   column is the audit copy and what the comparator reads.
 - **The work order:** `work_order_ref`, and `work_order` with the order itself

src/benchx/parquet.py

@@ -1,9 +1,10 @@
-"""The store's row: the message columns #30 derives, plus the store's own.
+"""The store's row: the message columns derived from the result schema, plus the store's own.
 
-`arrow_type` and `to_row` are the mapping of tools/json_to_parquet.py in #30
-(measurement-result-parquet.md §2): closed objects become structs, open
-objects canonical JSON, observations one typed list. The store adds the
-columns it derives at ingest beside them (prototype-design.md §4).
+`arrow_type` and `to_row` are the mapping of measurement-result-parquet.md §2:
+closed objects become structs, open objects canonical JSON, observations one
+typed list. `message_table` is that row alone (tools/json_to_parquet.py
+writes it); the store adds the columns it derives at ingest beside it
+(prototype-design.md §4).
 """
 
 from datetime import datetime
@@ -12,6 +13,7 @@
 import rfc8785
 
 from .core import catalog
+from .core.hashing import canonical, sha256_hex
 
 _VALIDATOR = catalog.lookup("measurement-result", catalog.MEASUREMENT_RESULT)
 if _VALIDATOR is None:
@@ -35,8 +37,15 @@ def arrow_type(node, defs, json_type):
     if "oneOf" in node:  # union of closed objects, or number-or-object observation
         branches = [defs[b["$ref"].rsplit("/", 1)[-1]] if "$ref" in b else b for b in node["oneOf"]]
         objects = [b for b in branches if b.get("type") == "object"]
-        if not objects:
-            raise ValueError(f"unsupported oneOf without an object alternative: {node}")
+        others = [b for b in branches if b.get("type") != "object"]
+        closed = all(b.get("additionalProperties") is False and "properties" in b for b in objects)
+        # to_row stores a bare value in `value`, so the only other alternative
+        # allowed is a number beside one object whose `value` is a number.
+        bare = [b.get("type") for b in others] == ["number"] and len(objects) == 1
+        if not objects or not closed or (others and not bare):
+            raise ValueError(f"unsupported oneOf: {node}")
+        if others and nested(objects[0]["properties"].get("value", {})) != SCALARS["number"]:
+            raise ValueError(f"unsupported oneOf: a bare number needs a number `value` to be stored in: {node}")
         required = set.intersection(*(set(b.get("required", [])) for b in objects))
         types = {}  # in first-appearance order
         for b in objects:
@@ -86,7 +95,7 @@ def to_row(data_type, value):
     pa.field("reported_coordinates_fingerprint", pa.string(), False),
     pa.field("payload_sha256", pa.string(), False),
     pa.field("attempt_signature", pa.string(), False),  # schema §5.3 attempt integrity
-    # The document itself, canonical: #30's columns cannot reproduce it (a bare
+    # The document itself, canonical: the message columns cannot reproduce it (a bare
     # number and a structured observation map alike), and the comparator reads it.
     pa.field("document", pa.json_(), False),
     pa.field("work_order_ref", pa.string()),
@@ -96,16 +105,29 @@ def to_row(data_type, value):
 _DEFS = RESULT_SCHEMA["$defs"]
 _BUILD = arrow_type(RESULT_SCHEMA, _DEFS, pa.string())
 _FINAL = arrow_type(RESULT_SCHEMA, _DEFS, pa.json_())
-RESULTS_SCHEMA = pa.schema(list(_FINAL) + DERIVED, metadata={
+# Which message schema a file was written under; the checksum tells drafts
+# apart that change under one URI (measurement-result-parquet.md §3).
+_METADATA = {
     "benchx.schema_uri": RESULT_SCHEMA["$id"],
-})
+    "benchx.message_schema_sha256": sha256_hex(canonical(RESULT_SCHEMA)),
+}
+MESSAGE_SCHEMA = pa.schema(list(_FINAL), metadata=_METADATA)
+RESULTS_SCHEMA = pa.schema(list(_FINAL) + DERIVED, metadata=_METADATA)
 
 
-def results_table(documents: list[dict], derived: list[dict]) -> pa.Table:
-    """One row per result: the message as #30 maps it, then the derived columns."""
-    rows = [{**to_row(_FINAL, doc), **extra} for doc, extra in zip(documents, derived)]
+def _table(rows: list[dict], extra: list[pa.Field], schema: pa.Schema) -> pa.Table:
     # Build with strings, then cast: PyArrow cannot fill nested JSON fields
     # from Python values directly.
-    build = [pa.field(f.name, pa.string(), f.nullable) if f.type == pa.json_() else f for f in DERIVED]
-    table = pa.Table.from_pylist(rows, schema=pa.schema(list(_BUILD) + build))
-    return table.cast(RESULTS_SCHEMA)
+    build = [pa.field(f.name, pa.string(), f.nullable) if f.type == pa.json_() else f for f in extra]
+    return pa.Table.from_pylist(rows, schema=pa.schema(list(_BUILD) + build)).cast(schema)
+
+
+def message_table(documents: list[dict]) -> pa.Table:
+    """One row per validated result: the message columns alone."""
+    return _table([to_row(_FINAL, doc) for doc in documents], [], MESSAGE_SCHEMA)
+
+
+def results_table(documents: list[dict], derived: list[dict]) -> pa.Table:
+    """One row per result: the message columns, then the derived columns."""
+    rows = [{**to_row(_FINAL, doc), **extra} for doc, extra in zip(documents, derived)]
+    return _table(rows, DERIVED, RESULTS_SCHEMA)

src/benchx/store.py

@@ -1,8 +1,8 @@
 """The local store: one Parquet file (prototype-design.md §4).
 
-One row per result: the message columns #30 derives from the result schema,
-then the columns the store derives, including the canonical document itself
-and the work order it references. Entities, series, and series points are
+One row per result: the message columns derived from the result schema
+(measurement-result-parquet.md), then the columns the store derives,
+including the canonical document itself and the work order it references. Entities, series, and series points are
 computed from the rows when asked for; nothing else is stored. JSON is for
 documents in transit, never for the store.