From 1e6803011c8fe414e3b5c02b76e0957a8a6f045a Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Fri, 18 Sep 2026 19:36:49 +0200 Subject: [PATCH 1/6] Add Parquet storage schema and a PyArrow example Describe the schema under which a validated measurement result is stored in Parquet, and add a small program that derives the Arrow schema from the message JSON Schema and writes messages to one Parquet file. --- README.md | 2 + docs/design/measurement-result-parquet.md | 155 ++++++++++++++++++++++ tools/json_to_parquet.py | 97 ++++++++++++++ 3 files changed, 254 insertions(+) create mode 100644 docs/design/measurement-result-parquet.md create mode 100644 tools/json_to_parquet.py diff --git a/README.md b/README.md index 8b35acc..6d21cfd 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,8 @@ 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, with a small PyArrow example 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 diff --git a/docs/design/measurement-result-parquet.md b/docs/design/measurement-result-parquet.md new file mode 100644 index 0000000..dbe0dec --- /dev/null +++ b/docs/design/measurement-result-parquet.md @@ -0,0 +1,155 @@ +# Measurement Results in Parquet + +**Status:** Draft for review; exploratory example + +**Companion to:** `benchmark-result-schema.md` §5 (the message; where this document and the schema differ, the schema wins) + +**Author:** Rok Mihevc + +## 1. Scope + +This document shows the schema under which a validated measurement result (§5.2 ingest object) is stored in a Parquet file, and a small PyArrow program that writes one: [`tools/json_to_parquet.py`](../../tools/json_to_parquet.py). It defines nothing about the message itself. + +One row is one result, that is one attempt and one quantity. A row holds the reported facts only. Series fingerprints, derived estimates, and entity tables are computed by a store at ingest (§4.3, §5.3) and would sit beside these columns; they are not part of a message and so not part of this schema. The mapping is not byte-preserving, so a store that needs the original document keeps it separately. + +## 2. Mapping + +The Arrow schema is derived from [`schema.json`](../../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 | +| `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` | + +Open objects stay JSON because their keys are project-defined: workload parameters, subject configuration, comparison context, environment identity, resource selection, observed context, 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; `*` marks a non-nullable field. + +```text +schema_version*: int64 +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 + quantity*: struct + name*: string, unit*: string, direction: string, deterministic: bool + comparison_context*: json + environment*: struct + schema*: string, identity*: json, metadata: json + resource_selection: json +measurement*: struct + 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 +observed_context: json, 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 = +``` + +The checksum matters while 0.1.0 is a draft that changes under one URI. + +## 4. Writing and reading + +The program validates each message against the JSON Schema, derives the Arrow schema, and writes all messages given to it as the rows of one file: + +```bash +pip install jsonschema pyarrow rfc8785 +python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ + results.parquet schemas/measurement-result/0.1.0/examples/*.json +``` + +Any Arrow-capable engine reads the result directly. 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, + coordinates.quantity.name AS quantity, + len(measurement.observations) AS n, + list_min([o.value FOR o IN measurement.observations]) AS fastest, + coordinates.comparison_context->>'$.protocol.timer' AS timer +FROM 'results.parquet' +ORDER BY workload, quantity; +``` + +```text +workload quantity n fastest timer +matmul cpu-time 10 0.000212 perf_counter +matmul gpu-time 10 0.01836 cuda-events +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 overhead ratio is an aggregate-only compound result, so it has no observations. + +## 5. 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. +- Derived columns and tables, appending to a dataset, partitioning, and compaction. +- Reading a message back out of Parquet. Member order, number spelling, and whether an observation was a bare number are not kept. +- Timestamps finer than microseconds are truncated to the column's precision. diff --git a/tools/json_to_parquet.py b/tools/json_to_parquet.py new file mode 100644 index 0000000..8546c7e --- /dev/null +++ b/tools/json_to_parquet.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Store measurement-result JSON messages in a Parquet file with PyArrow. + +The Arrow schema is derived from the message JSON Schema, so it cannot drift: +closed objects become structs, open objects canonical JSON strings, and an +observation, bare number or structured, one struct. See +docs/design/measurement-result-parquet.md. + + pip install jsonschema pyarrow rfc8785 + python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ + results.parquet schemas/measurement-result/0.1.0/examples/*.json +""" + +import hashlib +import json +import sys +from datetime import datetime + +import jsonschema +import pyarrow as pa +import pyarrow.parquet as pq +import rfc8785 + +TIMESTAMP = pa.timestamp("us", tz="UTC") +SCALARS = {"string": pa.string(), "integer": pa.int64(), "number": pa.float64(), "boolean": pa.bool_()} + + +def arrow_type(node, defs, json_type): + """Map one JSON Schema node to an Arrow type.""" + if "$ref" in node: + node = defs[node["$ref"].rsplit("/", 1)[-1]] + nested = lambda n: arrow_type(n, defs, json_type) # noqa: E731 + if "allOf" in node and "type" not in node: # a reference narrowed by assertions + return nested(node["allOf"][0]) + 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"] + required = set.intersection(*(set(b.get("required", [])) for b in objects)) + properties = {k: v for b in reversed(objects) for k, v in b["properties"].items()} + order = list(dict.fromkeys(k for b in objects for k in b["properties"])) + return pa.struct([pa.field(k, nested(properties[k]), k not in required) for k in order]) + if "const" in node or "enum" in node: + return pa.int64() if isinstance(node.get("const"), int) else pa.string() + if node.get("type") == "array": + return pa.list_(pa.field("element", nested(node["items"]), False)) + if node.get("type") == "object": + if node.get("additionalProperties") is not False or "properties" not in node: + return json_type + required = node.get("required", []) + return pa.struct([pa.field(k, nested(v), k not in required) for k, v in node["properties"].items()]) + if node.get("format") == "date-time": + return TIMESTAMP + scalar = node.get("type") + return SCALARS[scalar] if isinstance(scalar, str) else json_type # mixed primitives stay JSON + + +def to_row(data_type, value): + """Shape one message value for its Arrow type.""" + if value is None: + return None + if pa.types.is_struct(data_type): + if not isinstance(value, dict): # a bare observation number + value = {"value": value} + return {f.name: to_row(f.type, value.get(f.name)) for f in data_type} + if pa.types.is_list(data_type): + return [to_row(data_type.value_type, v) for v in value] + if data_type == pa.json_(): # open object as canonical JSON + return rfc8785.dumps(value).decode() + if data_type == TIMESTAMP: + return datetime.fromisoformat(value) + return value + + +def main(schema_path, output, *message_paths): + with open(schema_path) as f: + message_schema = json.load(f) + messages = [] + for path in message_paths: + with open(path) as f: + messages.append(json.load(f)) + jsonschema.validate(messages[-1], message_schema) + + defs = message_schema["$defs"] + # Build with strings, then cast: PyArrow cannot fill nested JSON fields + # from Python values directly. + build = arrow_type(message_schema, defs, pa.string()) + final = arrow_type(message_schema, defs, pa.json_()) + table = pa.Table.from_pylist([to_row(pa.struct(final), m) for m in messages], schema=pa.schema(build)) + table = table.cast(pa.schema(final)).replace_schema_metadata({ + "benchx.schema_uri": message_schema["$id"], + "benchx.message_schema_sha256": hashlib.sha256(rfc8785.dumps(message_schema)).hexdigest(), + }) + pq.write_table(table, output, compression="zstd") + + +if __name__ == "__main__": + main(*sys.argv[1:]) From a438bdfa13b0784f9c261da56f87ce10bafabaad Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Fri, 18 Sep 2026 23:18:59 +0200 Subject: [PATCH 2/6] Read messages strictly before storing them Reject duplicate object keys, non-finite number tokens, and numbers outside the RFC 8785 domain, including integers beyond the I-JSON safe range, before JSON Schema validation. Document why in the Parquet notes. --- docs/design/measurement-result-parquet.md | 10 ++++++- tools/json_to_parquet.py | 32 ++++++++++++++++++----- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/docs/design/measurement-result-parquet.md b/docs/design/measurement-result-parquet.md index dbe0dec..df4a338 100644 --- a/docs/design/measurement-result-parquet.md +++ b/docs/design/measurement-result-parquet.md @@ -116,7 +116,7 @@ The checksum matters while 0.1.0 is a draft that changes under one URI. ## 4. Writing and reading -The program validates each message against the JSON Schema, derives the Arrow schema, and writes all messages given to it as the rows of one file: +The program reads each message strictly, validates it against the JSON Schema, derives the Arrow schema, and writes all messages given to it as the rows of one file: ```bash pip install jsonschema pyarrow rfc8785 @@ -124,6 +124,14 @@ python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ results.parquet 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 the program 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. +- **Numbers outside the RFC 8785 domain.** Every message is passed through RFC 8785 serialization, which refuses integers beyond ±2^53, the I-JSON safe-integer range. A larger value is read differently by parsers that hold numbers as binary64, so it would not mean the same thing to every reader. + +The same canonical form is what the `json` columns hold and what the schema checksum is computed over. One rejected message fails the run and nothing is written. + Any Arrow-capable engine reads the result directly. In DuckDB, structs are addressed with dots, the observation list with list functions, and JSON columns with JSON paths: ```sql diff --git a/tools/json_to_parquet.py b/tools/json_to_parquet.py index 8546c7e..9825bd1 100644 --- a/tools/json_to_parquet.py +++ b/tools/json_to_parquet.py @@ -71,14 +71,32 @@ def to_row(data_type, value): return value +def unique_keys(pairs): + # jsonschema only sees the parsed mapping, after duplicate keys are lost. + result = dict(pairs) + if len(result) != len(pairs): + raise ValueError(f"duplicate JSON key in {[key for key, _ in pairs]}") + return result + + +def non_finite(token): + # Python's JSON parser accepts NaN and Infinity even though JSON does not. + raise ValueError(f"invalid JSON number: {token}") + + +def load_json(path): + """Load strict I-JSON: UTF-8, unique keys, finite numbers, safe integers.""" + with open(path, encoding="utf-8") as f: + value = json.load(f, object_pairs_hook=unique_keys, parse_constant=non_finite) + rfc8785.dumps(value) # rejects numbers outside the RFC 8785 domain, such as integers beyond 2**53 + return value + + def main(schema_path, output, *message_paths): - with open(schema_path) as f: - message_schema = json.load(f) - messages = [] - for path in message_paths: - with open(path) as f: - messages.append(json.load(f)) - jsonschema.validate(messages[-1], message_schema) + message_schema = load_json(schema_path) + messages = [load_json(path) for path in message_paths] + for message in messages: + jsonschema.validate(message, message_schema) defs = message_schema["$defs"] # Build with strings, then cast: PyArrow cannot fill nested JSON fields From dd6df90c184e52995c65040948abaa02b1393937 Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Thu, 24 Sep 2026 23:59:33 +0200 Subject: [PATCH 3/6] Fix Parquet type mapping, add tests, and record per-observation iterations --- docs/design/measurement-result-parquet.md | 49 +++++-- .../measurement-result/0.1.0/schema.json | 5 + tests/test_json_to_parquet.py | 127 ++++++++++++++++++ tools/json_to_parquet.py | 35 ++++- 4 files changed, 199 insertions(+), 17 deletions(-) create mode 100644 tests/test_json_to_parquet.py diff --git a/docs/design/measurement-result-parquet.md b/docs/design/measurement-result-parquet.md index df4a338..2e6aa12 100644 --- a/docs/design/measurement-result-parquet.md +++ b/docs/design/measurement-result-parquet.md @@ -2,7 +2,7 @@ **Status:** Draft for review; exploratory example -**Companion to:** `benchmark-result-schema.md` §5 (the message; where this document and the schema differ, the schema wins) +**Companion to:** `benchmark-result-schema.md` §5 (the message; where this document and the schema differ, the schema wins), `prototype-design.md` §4 (the prototype store whose `results/` table this row defines) **Author:** Rok Mihevc @@ -10,7 +10,7 @@ This document shows the schema under which a validated measurement result (§5.2 ingest object) is stored in a Parquet file, and a small PyArrow program that writes one: [`tools/json_to_parquet.py`](../../tools/json_to_parquet.py). It defines nothing about the message itself. -One row is one result, that is one attempt and one quantity. A row holds the reported facts only. Series fingerprints, derived estimates, and entity tables are computed by a store at ingest (§4.3, §5.3) and would sit beside these columns; they are not part of a message and so not part of this schema. The mapping is not byte-preserving, so a store that needs the original document keeps it separately. +One row is one result, that is one attempt and one quantity. A row holds the reported facts only. Series fingerprints, derived estimates, and entity tables are computed by a store at ingest (§4.3, §5.3) and sit beside these columns (§4 below); they are not part of a message. The mapping is not byte-preserving, so a store that needs the original document keeps it separately, as the prototype store does under `raw/`. ## 2. Mapping @@ -20,13 +20,15 @@ The Arrow schema is derived from [`schema.json`](../../schemas/measurement-resul |---|---| | 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 | +| `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` | +| `const` or `enum` | by its values: all strings `string`, all integers `int64`, other numbers `float64`, all booleans `bool`, mixed `json` | +| `oneOf` of primitives only | `json` | +| string | `string` | +| integer | `int64` | | number | `float64` | | boolean | `bool` | @@ -34,7 +36,7 @@ Open objects stay JSON because their keys are project-defined: workload paramete ## 3. Resulting schema -For message schema 0.1.0; `*` marks a non-nullable field. +For message schema 0.1.0, whose messages carry `schema_version: 5`: the file name and `$id` version the schema document, the integer versions the message contract. `*` marks a non-nullable field. ```text schema_version*: int64 @@ -70,7 +72,8 @@ measurement*: 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], + value*: float64, ordinal: int64, inner_iterations: int64, slot: int64, + time: timestamp[us, UTC], group: string, pair: string, included: bool, exclusion_reason: string derivation: struct (as estimator) summaries: list of struct @@ -114,12 +117,30 @@ benchx.message_schema_sha256 = The checksum matters while 0.1.0 is a draft that changes under one URI. -## 4. Writing and reading +## 4. Columns a store adds + +The prototype store (`prototype-design.md` §4) writes this row as its `results/` table and adds, beside the message columns and outside the `coordinates`, `measurement`, and `provenance` structs, the columns it derives at ingest: + +```text +reported_coordinates_fingerprint*: string schema §4.3 +series_fingerprint: string per identity-policy schema; absent for thin results +estimate: float64, estimate_source: string the series point's value and whether producer or derived +payload_sha256*: string of the RFC 8785 form, for idempotency checks +ingested_at*: timestamp[us, UTC] +round: int64, slot: int64 promoted from procedure +inner_iterations: int64 promoted from procedure +``` + +The last three are promoted because the comparator's run mode pairs attempts by `procedure.round` and checks the schedule by `procedure.slot` (schema §5.5), and per-iteration values need the loop count; reading them from a JSON column would parse it on every row of every comparison. They are copies: the `procedure` column keeps the message's value, and a promoted column is rebuilt from it. Other `procedure` or `comparison_context` keys are promoted the same way when a query needs them, never by changing the message schema. + +**Appending and schema changes.** A store appends one file per ingest call. Because the Arrow schema is derived, a change to `schema.json` changes the file schema, and a dataset then holds files written under several message schemas, told apart by `benchx.message_schema_sha256`. Additive changes, a new optional field such as `inner_iterations` on observations, are read together with PyArrow's permissive schema unification (`pyarrow.dataset` with `unify_schemas(..., promote_options="permissive")`), where missing fields read as null. A change that removes a field or changes its type is a new message contract version (`schema_version`) and is written to a separate dataset or rewritten on migration, never mixed. + +## 5. Writing and reading The program reads each message strictly, validates it against the JSON Schema, derives the Arrow schema, and writes all messages given to it as the rows of one file: ```bash -pip install jsonschema pyarrow rfc8785 +pip install jsonschema 'pyarrow>=19' rfc8785 # Python 3.11 or newer python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ results.parquet schemas/measurement-result/0.1.0/examples/*.json ``` @@ -155,9 +176,15 @@ truncate wall-time 5 0.000128 perf_counter The overhead ratio is an aggregate-only compound result, so it has no observations. -## 5. Not covered +`tests/test_json_to_parquet.py` converts the examples and checks row count, nullability, canonical JSON, observation values, file metadata, strict input, and the type-mapping rules: + +```bash +python -m unittest tests/test_json_to_parquet.py +``` + +## 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. -- Derived columns and tables, appending to a dataset, partitioning, and compaction. +- Entity tables, partitioning, and compaction; the prototype store rewrites its small entity tables and compacts on request. - Reading a message back out of Parquet. Member order, number spelling, and whether an observation was a bare number are not kept. - Timestamps finer than microseconds are truncated to the column's precision. diff --git a/src/benchx/schemas/measurement-result/0.1.0/schema.json b/src/benchx/schemas/measurement-result/0.1.0/schema.json index 414ea50..d6b0f25 100644 --- a/src/benchx/schemas/measurement-result/0.1.0/schema.json +++ b/src/benchx/schemas/measurement-result/0.1.0/schema.json @@ -574,6 +574,11 @@ "properties": { "value": { "$ref": "#/$defs/finiteNumber" }, "ordinal": { "type": "integer", "minimum": 0 }, + "inner_iterations": { + "description": "Inner iterations this value was normalized by, when the count differs between observations of one batch; a count shared by the whole batch is procedure.inner_iterations.", + "type": "integer", + "minimum": 1 + }, "slot": { "description": "Position of this observation in the run's realized order across all sides, for observation-level interleaving.", "type": "integer", diff --git a/tests/test_json_to_parquet.py b/tests/test_json_to_parquet.py new file mode 100644 index 0000000..f4e495e --- /dev/null +++ b/tests/test_json_to_parquet.py @@ -0,0 +1,127 @@ +"""Tests for tools/json_to_parquet.py. + + pip install jsonschema 'pyarrow>=19' rfc8785 + python -m unittest tests/test_json_to_parquet.py +""" + +import json +import sys +import tempfile +import unittest +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq +import rfc8785 + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "tools")) + +import json_to_parquet # noqa: E402 +from json_to_parquet import arrow_type, load_json, main # noqa: E402 + +SCHEMA_PATH = ROOT / "schemas/measurement-result/0.1.0/schema.json" +EXAMPLES = sorted((ROOT / "schemas/measurement-result/0.1.0/examples").glob("*.json")) + + +class ConversionTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = tempfile.TemporaryDirectory() + cls.output = Path(cls.tmp.name) / "results.parquet" + main(SCHEMA_PATH, cls.output, *EXAMPLES) + cls.table = pq.read_table(cls.output) + cls.messages = [load_json(path) for path in EXAMPLES] + + @classmethod + def tearDownClass(cls): + cls.tmp.cleanup() + + def test_one_row_per_message(self): + self.assertEqual(self.table.num_rows, len(EXAMPLES)) + self.assertEqual(self.table.column("ingest_key").to_pylist(), [m["ingest_key"] for m in self.messages]) + + def test_required_fields_are_not_nullable(self): + schema = self.table.schema + self.assertFalse(schema.field("ingest_key").nullable) + self.assertTrue(schema.field("project").nullable) + observation = schema.field("measurement").type.field("observations").type.value_type + self.assertFalse(observation.field("value").nullable) + self.assertTrue(observation.field("inner_iterations").nullable) + + def test_open_objects_are_canonical_json(self): + self.assertEqual(self.table.schema.field("procedure").type, pa.json_()) + for message, stored in zip(self.messages, self.table.column("procedure").to_pylist()): + expected = rfc8785.dumps(message["procedure"]).decode() if "procedure" in message else None + self.assertEqual(stored, expected) + + def test_observations_keep_their_values(self): + for message, stored in zip(self.messages, self.table.column("measurement").to_pylist()): + expected = message["measurement"].get("observations") + if expected is None: + self.assertIsNone(stored["observations"]) + continue + values = [o["value"] if isinstance(o, dict) else o for o in expected] + self.assertEqual([o["value"] for o in stored["observations"]], values) + + def test_file_metadata_names_the_message_schema(self): + metadata = self.table.schema.metadata + self.assertEqual(metadata[b"benchx.schema_uri"], b"urn:benchx:schema:measurement-result:0.1.0") + self.assertEqual(len(metadata[b"benchx.message_schema_sha256"]), 64) + + +class StrictInputTests(unittest.TestCase): + def load(self, text): + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: + f.write(text) + try: + return load_json(f.name) + finally: + Path(f.name).unlink() + + def test_rejects_duplicate_keys(self): + with self.assertRaises(ValueError): + self.load('{"a": 1, "a": 2}') + + def test_rejects_non_finite_numbers(self): + for token in ("NaN", "Infinity", "-Infinity"): + with self.assertRaises(ValueError): + self.load(f'{{"a": {token}}}') + + def test_rejects_unsafe_integers(self): + with self.assertRaises(Exception): + self.load(json.dumps({"a": 2**53 + 1})) + + +class TypeMappingTests(unittest.TestCase): + def map(self, node, defs=None): + return arrow_type(node, defs or {}, pa.json_()) + + def test_literals(self): + self.assertEqual(self.map({"const": 5}), pa.int64()) + self.assertEqual(self.map({"const": True}), pa.bool_()) + self.assertEqual(self.map({"enum": [1, 2]}), pa.int64()) + self.assertEqual(self.map({"enum": [1, 2.5]}), pa.float64()) + self.assertEqual(self.map({"enum": ["a", "b"]}), pa.string()) + self.assertEqual(self.map({"enum": ["a", 1]}), pa.json_()) + + def test_union_of_primitives_stays_json(self): + self.assertEqual(self.map({"oneOf": [{"type": "string"}, {"type": "integer"}]}), pa.json_()) + + def test_conflicting_union_fails(self): + node = {"oneOf": [ + {"type": "object", "additionalProperties": False, "properties": {"x": {"type": "string"}}}, + {"type": "object", "additionalProperties": False, "properties": {"x": {"type": "integer"}}}, + ]} + with self.assertRaises(ValueError): + self.map(node) + + def test_every_def_maps(self): + schema = json_to_parquet.load_json(SCHEMA_PATH) + for name, node in schema["$defs"].items(): + with self.subTest(name=name): + self.map(node, schema["$defs"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/json_to_parquet.py b/tools/json_to_parquet.py index 9825bd1..096d576 100644 --- a/tools/json_to_parquet.py +++ b/tools/json_to_parquet.py @@ -6,7 +6,7 @@ observation, bare number or structured, one struct. See docs/design/measurement-result-parquet.md. - pip install jsonschema pyarrow rfc8785 + pip install jsonschema 'pyarrow>=19' rfc8785 # Python 3.11 or newer python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ results.parquet schemas/measurement-result/0.1.0/examples/*.json """ @@ -25,6 +25,19 @@ SCALARS = {"string": pa.string(), "integer": pa.int64(), "number": pa.float64(), "boolean": pa.bool_()} +def literal_type(values, json_type): + """Map the values of a const or enum to one Arrow type.""" + if all(isinstance(v, bool) for v in values): + return pa.bool_() + if all(isinstance(v, int) and not isinstance(v, bool) for v in values): + return pa.int64() + if all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values): + return pa.float64() + if all(isinstance(v, str) for v in values): + return pa.string() + return json_type # mixed literals stay JSON + + def arrow_type(node, defs, json_type): """Map one JSON Schema node to an Arrow type.""" if "$ref" in node: @@ -35,12 +48,20 @@ 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: # a union of primitives + return json_type required = set.intersection(*(set(b.get("required", [])) for b in objects)) - properties = {k: v for b in reversed(objects) for k, v in b["properties"].items()} - order = list(dict.fromkeys(k for b in objects for k in b["properties"])) - return pa.struct([pa.field(k, nested(properties[k]), k not in required) for k in order]) - if "const" in node or "enum" in node: - return pa.int64() if isinstance(node.get("const"), int) else pa.string() + types = {} # in first-appearance order + for b in objects: + for k, v in b["properties"].items(): + t = nested(v) + if types.setdefault(k, t) != t: + raise ValueError(f"oneOf alternatives disagree on the type of {k!r}: {types[k]} and {t}") + return pa.struct([pa.field(k, t, k not in required) for k, t in types.items()]) + if "const" in node: + return literal_type([node["const"]], json_type) + if "enum" in node: + return literal_type(node["enum"], json_type) if node.get("type") == "array": return pa.list_(pa.field("element", nested(node["items"]), False)) if node.get("type") == "object": @@ -112,4 +133,6 @@ def main(schema_path, output, *message_paths): if __name__ == "__main__": + if len(sys.argv) < 4: + sys.exit("usage: json_to_parquet.py SCHEMA OUTPUT MESSAGE...") main(*sys.argv[1:]) From 58d7ab2f73d5956e9742d506a3b1a9f3897b9e5d Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Fri, 25 Sep 2026 00:05:47 +0200 Subject: [PATCH 4/6] Minimize Parquet PR: move schema change out, trim converter and tests --- docs/design/measurement-result-parquet.md | 28 ++--- .../measurement-result/0.1.0/schema.json | 5 - tests/test_json_to_parquet.py | 103 +++++------------- tools/json_to_parquet.py | 28 ++--- 4 files changed, 43 insertions(+), 121 deletions(-) diff --git a/docs/design/measurement-result-parquet.md b/docs/design/measurement-result-parquet.md index 2e6aa12..3ca9f32 100644 --- a/docs/design/measurement-result-parquet.md +++ b/docs/design/measurement-result-parquet.md @@ -25,12 +25,11 @@ The Arrow schema is derived from [`schema.json`](../../schemas/measurement-resul | `allOf` of a reference and extra assertions (`derivation`) | the referenced type | | array | `list` | | string with `format: date-time` | `timestamp[us, UTC]` | -| `const` or `enum` | by its values: all strings `string`, all integers `int64`, other numbers `float64`, all booleans `bool`, mixed `json` | -| `oneOf` of primitives only | `json` | -| string | `string` | -| integer | `int64` | +| 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, comparison context, environment identity, resource selection, observed context, and procedure. Canonical text makes equal values equal strings. A store promotes a key to a typed column only when a query needs it. @@ -72,8 +71,7 @@ measurement*: struct kind*: string, lower: float64, upper: float64, lower_inclusive: bool, upper_inclusive: bool, cause*: string observations: list of struct - value*: float64, ordinal: int64, inner_iterations: int64, slot: int64, - time: timestamp[us, UTC], + 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 @@ -117,23 +115,11 @@ benchx.message_schema_sha256 = The checksum matters while 0.1.0 is a draft that changes under one URI. -## 4. Columns a store adds +## 4. Beside the message columns -The prototype store (`prototype-design.md` §4) writes this row as its `results/` table and adds, beside the message columns and outside the `coordinates`, `measurement`, and `provenance` structs, the columns it derives at ingest: +A store adds the columns it derives at ingest, such as fingerprints, the series point's estimate, and the ingest time, and may copy a key out of an open object into a typed column when a query needs it, `procedure.round` for the comparator's run mode for instance. Those columns and the store's layout belong to the store's design (`prototype-design.md` §4), not to this document. -```text -reported_coordinates_fingerprint*: string schema §4.3 -series_fingerprint: string per identity-policy schema; absent for thin results -estimate: float64, estimate_source: string the series point's value and whether producer or derived -payload_sha256*: string of the RFC 8785 form, for idempotency checks -ingested_at*: timestamp[us, UTC] -round: int64, slot: int64 promoted from procedure -inner_iterations: int64 promoted from procedure -``` - -The last three are promoted because the comparator's run mode pairs attempts by `procedure.round` and checks the schedule by `procedure.slot` (schema §5.5), and per-iteration values need the loop count; reading them from a JSON column would parse it on every row of every comparison. They are copies: the `procedure` column keeps the message's value, and a promoted column is rebuilt from it. Other `procedure` or `comparison_context` keys are promoted the same way when a query needs them, never by changing the message schema. - -**Appending and schema changes.** A store appends one file per ingest call. Because the Arrow schema is derived, a change to `schema.json` changes the file schema, and a dataset then holds files written under several message schemas, told apart by `benchx.message_schema_sha256`. Additive changes, a new optional field such as `inner_iterations` on observations, are read together with PyArrow's permissive schema unification (`pyarrow.dataset` with `unify_schemas(..., promote_options="permissive")`), where missing fields read as null. A change that removes a field or changes its type is a new message contract version (`schema_version`) and is written to a separate dataset or rewritten on migration, never mixed. +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 files under PyArrow's permissive schema unification; removing a field or changing its type is a new `schema_version` and is not mixed into one dataset. ## 5. Writing and reading diff --git a/src/benchx/schemas/measurement-result/0.1.0/schema.json b/src/benchx/schemas/measurement-result/0.1.0/schema.json index d6b0f25..414ea50 100644 --- a/src/benchx/schemas/measurement-result/0.1.0/schema.json +++ b/src/benchx/schemas/measurement-result/0.1.0/schema.json @@ -574,11 +574,6 @@ "properties": { "value": { "$ref": "#/$defs/finiteNumber" }, "ordinal": { "type": "integer", "minimum": 0 }, - "inner_iterations": { - "description": "Inner iterations this value was normalized by, when the count differs between observations of one batch; a count shared by the whole batch is procedure.inner_iterations.", - "type": "integer", - "minimum": 1 - }, "slot": { "description": "Position of this observation in the run's realized order across all sides, for observation-level interleaving.", "type": "integer", diff --git a/tests/test_json_to_parquet.py b/tests/test_json_to_parquet.py index f4e495e..b06d81d 100644 --- a/tests/test_json_to_parquet.py +++ b/tests/test_json_to_parquet.py @@ -17,7 +17,6 @@ ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "tools")) -import json_to_parquet # noqa: E402 from json_to_parquet import arrow_type, load_json, main # noqa: E402 SCHEMA_PATH = ROOT / "schemas/measurement-result/0.1.0/schema.json" @@ -27,27 +26,16 @@ class ConversionTests(unittest.TestCase): @classmethod def setUpClass(cls): - cls.tmp = tempfile.TemporaryDirectory() - cls.output = Path(cls.tmp.name) / "results.parquet" - main(SCHEMA_PATH, cls.output, *EXAMPLES) - cls.table = pq.read_table(cls.output) + with tempfile.TemporaryDirectory() as tmp: + output = Path(tmp) / "results.parquet" + main(SCHEMA_PATH, output, *EXAMPLES) + cls.table = pq.read_table(output) cls.messages = [load_json(path) for path in EXAMPLES] - @classmethod - def tearDownClass(cls): - cls.tmp.cleanup() - - def test_one_row_per_message(self): - self.assertEqual(self.table.num_rows, len(EXAMPLES)) + def test_rows_and_nullability(self): self.assertEqual(self.table.column("ingest_key").to_pylist(), [m["ingest_key"] for m in self.messages]) - - def test_required_fields_are_not_nullable(self): - schema = self.table.schema - self.assertFalse(schema.field("ingest_key").nullable) - self.assertTrue(schema.field("project").nullable) - observation = schema.field("measurement").type.field("observations").type.value_type - self.assertFalse(observation.field("value").nullable) - self.assertTrue(observation.field("inner_iterations").nullable) + self.assertFalse(self.table.schema.field("ingest_key").nullable) + self.assertTrue(self.table.schema.field("project").nullable) def test_open_objects_are_canonical_json(self): self.assertEqual(self.table.schema.field("procedure").type, pa.json_()) @@ -58,69 +46,36 @@ def test_open_objects_are_canonical_json(self): def test_observations_keep_their_values(self): for message, stored in zip(self.messages, self.table.column("measurement").to_pylist()): expected = message["measurement"].get("observations") - if expected is None: - self.assertIsNone(stored["observations"]) - continue - values = [o["value"] if isinstance(o, dict) else o for o in expected] - self.assertEqual([o["value"] for o in stored["observations"]], values) + values = None if expected is None else [o["value"] if isinstance(o, dict) else o for o in expected] + self.assertEqual(values, None if stored["observations"] is None else [o["value"] for o in stored["observations"]]) def test_file_metadata_names_the_message_schema(self): - metadata = self.table.schema.metadata - self.assertEqual(metadata[b"benchx.schema_uri"], b"urn:benchx:schema:measurement-result:0.1.0") - self.assertEqual(len(metadata[b"benchx.message_schema_sha256"]), 64) + self.assertEqual(self.table.schema.metadata[b"benchx.schema_uri"], b"urn:benchx:schema:measurement-result:0.1.0") class StrictInputTests(unittest.TestCase): - def load(self, text): - with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: - f.write(text) - try: - return load_json(f.name) - finally: - Path(f.name).unlink() - - def test_rejects_duplicate_keys(self): - with self.assertRaises(ValueError): - self.load('{"a": 1, "a": 2}') - - def test_rejects_non_finite_numbers(self): - for token in ("NaN", "Infinity", "-Infinity"): - with self.assertRaises(ValueError): - self.load(f'{{"a": {token}}}') - - def test_rejects_unsafe_integers(self): - with self.assertRaises(Exception): - self.load(json.dumps({"a": 2**53 + 1})) + # The message schema accepts NaN and integers beyond 2**53 once parsed, and + # never sees duplicate keys, so these checks are the loader's alone. + def test_rejects_what_the_schema_cannot_see(self): + for text in ('{"a": 1, "a": 2}', '{"a": NaN}', '{"a": Infinity}', json.dumps({"a": 2**53 + 1})): + with self.subTest(text=text), tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "m.json" + path.write_text(text) + with self.assertRaises(Exception): + load_json(path) class TypeMappingTests(unittest.TestCase): - def map(self, node, defs=None): - return arrow_type(node, defs or {}, pa.json_()) - - def test_literals(self): - self.assertEqual(self.map({"const": 5}), pa.int64()) - self.assertEqual(self.map({"const": True}), pa.bool_()) - self.assertEqual(self.map({"enum": [1, 2]}), pa.int64()) - self.assertEqual(self.map({"enum": [1, 2.5]}), pa.float64()) - self.assertEqual(self.map({"enum": ["a", "b"]}), pa.string()) - self.assertEqual(self.map({"enum": ["a", 1]}), pa.json_()) - - def test_union_of_primitives_stays_json(self): - self.assertEqual(self.map({"oneOf": [{"type": "string"}, {"type": "integer"}]}), pa.json_()) - - def test_conflicting_union_fails(self): - node = {"oneOf": [ - {"type": "object", "additionalProperties": False, "properties": {"x": {"type": "string"}}}, - {"type": "object", "additionalProperties": False, "properties": {"x": {"type": "integer"}}}, - ]} - with self.assertRaises(ValueError): - self.map(node) - - def test_every_def_maps(self): - schema = json_to_parquet.load_json(SCHEMA_PATH) - for name, node in schema["$defs"].items(): - with self.subTest(name=name): - self.map(node, schema["$defs"]) + def test_unsupported_nodes_fail(self): + closed = lambda t: {"type": "object", "additionalProperties": False, "properties": {"x": {"type": t}}} # noqa: E731 + for node in ( + {"const": True}, + {"enum": ["a", 1]}, + {"oneOf": [{"type": "string"}, {"type": "integer"}]}, + {"oneOf": [closed("string"), closed("integer")]}, + ): + with self.subTest(node=node), self.assertRaises(ValueError): + arrow_type(node, {}, pa.json_()) if __name__ == "__main__": diff --git a/tools/json_to_parquet.py b/tools/json_to_parquet.py index 096d576..024f53b 100644 --- a/tools/json_to_parquet.py +++ b/tools/json_to_parquet.py @@ -25,19 +25,6 @@ SCALARS = {"string": pa.string(), "integer": pa.int64(), "number": pa.float64(), "boolean": pa.bool_()} -def literal_type(values, json_type): - """Map the values of a const or enum to one Arrow type.""" - if all(isinstance(v, bool) for v in values): - return pa.bool_() - if all(isinstance(v, int) and not isinstance(v, bool) for v in values): - return pa.int64() - if all(isinstance(v, (int, float)) and not isinstance(v, bool) for v in values): - return pa.float64() - if all(isinstance(v, str) for v in values): - return pa.string() - return json_type # mixed literals stay JSON - - def arrow_type(node, defs, json_type): """Map one JSON Schema node to an Arrow type.""" if "$ref" in node: @@ -48,8 +35,8 @@ 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: # a union of primitives - return json_type + if not objects: + raise ValueError(f"unsupported oneOf without an object alternative: {node}") required = set.intersection(*(set(b.get("required", [])) for b in objects)) types = {} # in first-appearance order for b in objects: @@ -58,10 +45,11 @@ def arrow_type(node, defs, json_type): if types.setdefault(k, t) != t: raise ValueError(f"oneOf alternatives disagree on the type of {k!r}: {types[k]} and {t}") return pa.struct([pa.field(k, t, k not in required) for k, t in types.items()]) - if "const" in node: - return literal_type([node["const"]], json_type) - if "enum" in node: - return literal_type(node["enum"], json_type) + if "const" in node or "enum" in node: + literals = {type(v) for v in node.get("enum", [node.get("const")])} + if literals not in ({str}, {int}): # bool is not int here: type(True) is bool + raise ValueError(f"unsupported const or enum: {node}") + return pa.string() if literals == {str} else pa.int64() if node.get("type") == "array": return pa.list_(pa.field("element", nested(node["items"]), False)) if node.get("type") == "object": @@ -133,6 +121,4 @@ def main(schema_path, output, *message_paths): if __name__ == "__main__": - if len(sys.argv) < 4: - sys.exit("usage: json_to_parquet.py SCHEMA OUTPUT MESSAGE...") main(*sys.argv[1:]) From 6dc45d84dd75bc095dba5de2330d85bb538658db Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Wed, 11 Nov 2026 14:40:36 +0100 Subject: [PATCH 5/6] MAINT: bring the Parquet design up to date with main MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit main has since taken the mapping into src/benchx/parquet.py and moved the schemas into the package, and the result schema has changed under it. - benchx.parquet: add message_table() and MESSAGE_SCHEMA (the message columns alone), and record benchx.message_schema_sha256 in the file metadata of both the message table and the store, as the document says. - tools/json_to_parquet.py: a thin wrapper over benchx; it reads and checks documents as `bx validate` does and writes message_table(). - tests/test_parquet.py: pytest tests of benchx.parquet and the tool, replacing tests/test_json_to_parquet.py; strict reading is already covered by test_loader.py. - measurement-result-parquet.md: regenerate §3 for the current schema (string schema_version, quantity under measurement, protocol, system, measurement.kind), describe the store's derived columns and document column, use the shipped paths, and rerun the DuckDB query over the six examples. - List the document in the design index and the site nav, and replace the "#30" references in the code and docs with the document's name. --- README.md | 3 +- docs/benchx.md | 3 +- docs/design/README.md | 4 +- docs/design/measurement-result-parquet.md | 83 +++++++------- docs/design/prototype-design.md | 6 +- src/benchx/parquet.py | 43 ++++--- src/benchx/store.py | 6 +- tests/test_json_to_parquet.py | 82 ------------- tests/test_parquet.py | 102 ++++++++++++++++ tools/json_to_parquet.py | 134 ++++------------------ zensical.toml | 1 + 11 files changed, 209 insertions(+), 258 deletions(-) delete mode 100644 tests/test_json_to_parquet.py create mode 100644 tests/test_parquet.py diff --git a/README.md b/README.md index 6d21cfd..c3cea46 100644 --- a/README.md +++ b/README.md @@ -52,7 +52,8 @@ 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, with a small PyArrow example in `tools/json_to_parquet.py`. + 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 diff --git a/docs/benchx.md b/docs/benchx.md index b6ad371..fa63201 100644 --- a/docs/benchx.md +++ b/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 diff --git a/docs/design/README.md b/docs/design/README.md index 7431ef9..a587d32 100644 --- a/docs/design/README.md +++ b/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, diff --git a/docs/design/measurement-result-parquet.md b/docs/design/measurement-result-parquet.md index 3ca9f32..711f8c9 100644 --- a/docs/design/measurement-result-parquet.md +++ b/docs/design/measurement-result-parquet.md @@ -1,20 +1,20 @@ # Measurement Results in Parquet -**Status:** Draft for review; exploratory example +**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 prototype store whose `results/` table this row defines) +**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 shows the schema under which a validated measurement result (§5.2 ingest object) is stored in a Parquet file, and a small PyArrow program that writes one: [`tools/json_to_parquet.py`](../../tools/json_to_parquet.py). It defines nothing about the message itself. +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. A row holds the reported facts only. Series fingerprints, derived estimates, and entity tables are computed by a store at ingest (§4.3, §5.3) and sit beside these columns (§4 below); they are not part of a message. The mapping is not byte-preserving, so a store that needs the original document keeps it separately, as the prototype store does under `raw/`. +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 [`schema.json`](../../schemas/measurement-result/0.1.0/schema.json) by the rules below, so it follows the message schema without being maintained by hand. +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 | |---|---| @@ -31,14 +31,14 @@ The Arrow schema is derived from [`schema.json`](../../schemas/measurement-resul | 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, comparison context, environment identity, resource selection, observed context, and procedure. Canonical text makes equal values equal strings. A store promotes a key to a typed column only when a query needs it. +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: 5`: the file name and `$id` version the schema document, the integer versions the message contract. `*` marks a non-nullable field. +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*: int64 +schema_version*: string producer*: struct name*: string, version*: string, mapping_version*: string ingest_key*: string, attempt_key*: string, project: string @@ -48,7 +48,7 @@ revision*: struct key*: string, native_order: int64, native_time: timestamp[us, UTC] benchmark: struct source*: struct (as source) - revision*: struct (as revision) + revision: struct (as revision) coordinates*: struct workload*: struct name*: string, parameters: json @@ -59,14 +59,14 @@ coordinates*: struct components*: list of struct role*: string, source: string, revision: string configuration: json - quantity*: struct - name*: string, unit*: string, direction: string, deterministic: bool - comparison_context*: json + protocol*: json environment*: struct - schema*: string, identity*: json, metadata: json + schema_version*: string, identity*: json, system: json resource_selection: json measurement*: struct - status*: string, reason: string + 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 @@ -84,7 +84,7 @@ measurement*: struct value: float64, source*: string, name: string, method: string, statistic: string, level: float64, lower: float64, upper: float64, label: string, coverage_factor: float64 -observed_context: json, procedure: json +procedure: json provenance*: struct run_key: string, labels: json, native_id: string, started_at*: timestamp[us, UTC], ended_at: timestamp[us, UTC], @@ -117,60 +117,57 @@ 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, such as fingerprints, the series point's estimate, and the ingest time, and may copy a key out of an open object into a typed column when a query needs it, `procedure.round` for the comparator's run mode for instance. Those columns and the store's layout belong to the store's design (`prototype-design.md` §4), not to this document. +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 files under PyArrow's permissive schema unification; removing a field or changing its type is a new `schema_version` and is not mixed into one dataset. +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 message strictly, validates it against the JSON Schema, derives the Arrow schema, and writes all messages given to it as the rows of one file: +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 jsonschema 'pyarrow>=19' rfc8785 # Python 3.11 or newer -python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ - results.parquet schemas/measurement-result/0.1.0/examples/*.json +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 the program rejects what falls outside it before validating: +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. -- **Numbers outside the RFC 8785 domain.** Every message is passed through RFC 8785 serialization, which refuses integers beyond ±2^53, the I-JSON safe-integer range. A larger value is read differently by parsers that hold numbers as binary64, so it would not mean the same thing to every reader. +- **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. -The same canonical form is what the `json` columns hold and what the schema checksum is computed over. One rejected message fails the run and nothing is written. +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. In DuckDB, structs are addressed with dots, the observation list with list functions, and JSON columns with JSON paths: +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, - coordinates.quantity.name AS quantity, + measurement.quantity.name AS quantity, len(measurement.observations) AS n, - list_min([o.value FOR o IN measurement.observations]) AS fastest, - coordinates.comparison_context->>'$.protocol.timer' AS timer + 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 fastest timer -matmul cpu-time 10 0.000212 perf_counter -matmul gpu-time 10 0.01836 cuda-events -parquet-read wall-time 5 0.0363 perf_counter -truncate overhead-ratio NULL NULL perf_counter -truncate wall-time 5 0.000128 perf_counter +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 overhead ratio is an aggregate-only compound result, so it has no observations. +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_json_to_parquet.py` converts the examples and checks row count, nullability, canonical JSON, observation values, file metadata, strict input, and the type-mapping rules: - -```bash -python -m unittest tests/test_json_to_parquet.py -``` +`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. -- Entity tables, partitioning, and compaction; the prototype store rewrites its small entity tables and compacts on request. -- Reading a message back out of Parquet. Member order, number spelling, and whether an observation was a bare number are not kept. +- 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. diff --git a/docs/design/prototype-design.md b/docs/design/prototype-design.md index 04e267b..2cd8da7 100644 --- a/docs/design/prototype-design.md +++ b/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,7 +404,7 @@ 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 @@ -412,7 +412,7 @@ no JSON, no separate entity tables. Each row is one result, the `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 diff --git a/src/benchx/parquet.py b/src/benchx/parquet.py index 0950b5c..c2b48da 100644 --- a/src/benchx/parquet.py +++ b/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: @@ -86,7 +88,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 +98,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) diff --git a/src/benchx/store.py b/src/benchx/store.py index ab22116..a7e1fef 100644 --- a/src/benchx/store.py +++ b/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. diff --git a/tests/test_json_to_parquet.py b/tests/test_json_to_parquet.py deleted file mode 100644 index b06d81d..0000000 --- a/tests/test_json_to_parquet.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Tests for tools/json_to_parquet.py. - - pip install jsonschema 'pyarrow>=19' rfc8785 - python -m unittest tests/test_json_to_parquet.py -""" - -import json -import sys -import tempfile -import unittest -from pathlib import Path - -import pyarrow as pa -import pyarrow.parquet as pq -import rfc8785 - -ROOT = Path(__file__).resolve().parents[1] -sys.path.insert(0, str(ROOT / "tools")) - -from json_to_parquet import arrow_type, load_json, main # noqa: E402 - -SCHEMA_PATH = ROOT / "schemas/measurement-result/0.1.0/schema.json" -EXAMPLES = sorted((ROOT / "schemas/measurement-result/0.1.0/examples").glob("*.json")) - - -class ConversionTests(unittest.TestCase): - @classmethod - def setUpClass(cls): - with tempfile.TemporaryDirectory() as tmp: - output = Path(tmp) / "results.parquet" - main(SCHEMA_PATH, output, *EXAMPLES) - cls.table = pq.read_table(output) - cls.messages = [load_json(path) for path in EXAMPLES] - - def test_rows_and_nullability(self): - self.assertEqual(self.table.column("ingest_key").to_pylist(), [m["ingest_key"] for m in self.messages]) - self.assertFalse(self.table.schema.field("ingest_key").nullable) - self.assertTrue(self.table.schema.field("project").nullable) - - def test_open_objects_are_canonical_json(self): - self.assertEqual(self.table.schema.field("procedure").type, pa.json_()) - for message, stored in zip(self.messages, self.table.column("procedure").to_pylist()): - expected = rfc8785.dumps(message["procedure"]).decode() if "procedure" in message else None - self.assertEqual(stored, expected) - - def test_observations_keep_their_values(self): - for message, stored in zip(self.messages, self.table.column("measurement").to_pylist()): - expected = message["measurement"].get("observations") - values = None if expected is None else [o["value"] if isinstance(o, dict) else o for o in expected] - self.assertEqual(values, None if stored["observations"] is None else [o["value"] for o in stored["observations"]]) - - def test_file_metadata_names_the_message_schema(self): - self.assertEqual(self.table.schema.metadata[b"benchx.schema_uri"], b"urn:benchx:schema:measurement-result:0.1.0") - - -class StrictInputTests(unittest.TestCase): - # The message schema accepts NaN and integers beyond 2**53 once parsed, and - # never sees duplicate keys, so these checks are the loader's alone. - def test_rejects_what_the_schema_cannot_see(self): - for text in ('{"a": 1, "a": 2}', '{"a": NaN}', '{"a": Infinity}', json.dumps({"a": 2**53 + 1})): - with self.subTest(text=text), tempfile.TemporaryDirectory() as tmp: - path = Path(tmp) / "m.json" - path.write_text(text) - with self.assertRaises(Exception): - load_json(path) - - -class TypeMappingTests(unittest.TestCase): - def test_unsupported_nodes_fail(self): - closed = lambda t: {"type": "object", "additionalProperties": False, "properties": {"x": {"type": t}}} # noqa: E731 - for node in ( - {"const": True}, - {"enum": ["a", 1]}, - {"oneOf": [{"type": "string"}, {"type": "integer"}]}, - {"oneOf": [closed("string"), closed("integer")]}, - ): - with self.subTest(node=node), self.assertRaises(ValueError): - arrow_type(node, {}, pa.json_()) - - -if __name__ == "__main__": - unittest.main() diff --git a/tests/test_parquet.py b/tests/test_parquet.py new file mode 100644 index 0000000..221c463 --- /dev/null +++ b/tests/test_parquet.py @@ -0,0 +1,102 @@ +"""The message columns of a result (measurement-result-parquet.md) and tools/json_to_parquet.py.""" + +import hashlib +import importlib.util +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq +import pytest +import rfc8785 + +from benchx import parquet +from benchx.core import loader +from benchx.core.catalog import SCHEMAS + +EXAMPLES = sorted((SCHEMAS / "measurement-result" / "0.1.0" / "examples").glob("*.json")) +# The tool lives in the checkout, not the package, so an installed copy skips its tests. +TOOL = Path(__file__).resolve().parents[1] / "tools" / "json_to_parquet.py" + + +@pytest.fixture(scope="module") +def messages(): + return [loader.read(path)[0] for path in EXAMPLES] + + +@pytest.fixture(scope="module") +def table(messages): + return parquet.message_table(messages) + + +def test_rows_and_nullability(table, messages): + assert table.schema == parquet.MESSAGE_SCHEMA + assert table.column("ingest_key").to_pylist() == [m["ingest_key"] for m in messages] + assert not table.schema.field("ingest_key").nullable + assert table.schema.field("project").nullable + + +def test_open_objects_are_canonical_json(table, messages): + assert table.schema.field("procedure").type == pa.json_() + expected = [rfc8785.dumps(m["procedure"]).decode() if "procedure" in m else None for m in messages] + assert table.column("procedure").to_pylist() == expected + + +def test_observations_keep_their_values(table, messages): + for message, stored in zip(messages, table.column("measurement").to_pylist()): + reported = message["measurement"].get("observations") + values = None if reported is None else [o["value"] if isinstance(o, dict) else o for o in reported] + assert values == (None if stored["observations"] is None else [o["value"] for o in stored["observations"]]) + + +def test_metadata_names_the_message_schema(table): + metadata = table.schema.metadata + assert metadata[b"benchx.schema_uri"] == parquet.RESULT_SCHEMA["$id"].encode() + digest = hashlib.sha256(rfc8785.dumps(parquet.RESULT_SCHEMA)).hexdigest() + assert metadata[b"benchx.message_schema_sha256"] == digest.encode() + + +def test_store_row_begins_with_the_message_columns(): + message = list(parquet.MESSAGE_SCHEMA) + assert list(parquet.RESULTS_SCHEMA)[: len(message)] == message + assert parquet.RESULTS_SCHEMA.metadata == parquet.MESSAGE_SCHEMA.metadata + + +def closed(json_type): + return {"type": "object", "additionalProperties": False, "properties": {"x": {"type": json_type}}} + + +@pytest.mark.parametrize("node", [ + {"const": True}, + {"enum": ["a", 1]}, + {"oneOf": [{"type": "string"}, {"type": "integer"}]}, + {"oneOf": [closed("string"), closed("integer")]}, +]) +def test_unsupported_nodes_fail(node): + with pytest.raises(ValueError): + parquet.arrow_type(node, {}, pa.json_()) + + +@pytest.fixture +def tool(): + if not TOOL.exists(): + pytest.skip("tools/ is not part of an installed benchx") + spec = importlib.util.spec_from_file_location("json_to_parquet", TOOL) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_tool_writes_the_message_columns(tool, tmp_path, table): + output = tmp_path / "results.parquet" + assert tool.main(output, *EXAMPLES) == 0 + assert pq.read_table(output).equals(table, check_metadata=True) + + +def test_tool_writes_nothing_when_one_document_is_invalid(tool, tmp_path, capsys): + bad = tmp_path / "bad.json" + text = EXAMPLES[0].read_text() + bad.write_text(text[: text.rindex("}")] + ', "ingest_key": "again"}') # a duplicate key + output = tmp_path / "results.parquet" + assert tool.main(output, EXAMPLES[0], bad) == 1 + assert not output.exists() + assert "bad.json: malformed" in capsys.readouterr().err diff --git a/tools/json_to_parquet.py b/tools/json_to_parquet.py index 024f53b..fe24211 100644 --- a/tools/json_to_parquet.py +++ b/tools/json_to_parquet.py @@ -1,124 +1,40 @@ #!/usr/bin/env python3 -"""Store measurement-result JSON messages in a Parquet file with PyArrow. +"""Write measurement-result JSON documents as the rows of one Parquet file. -The Arrow schema is derived from the message JSON Schema, so it cannot drift: -closed objects become structs, open objects canonical JSON strings, and an -observation, bare number or structured, one struct. See +Each document is read strictly and checked as `bx validate` checks it; the +rows are the message columns of benchx.parquet, whose Arrow schema is derived +from the shipped result schema, so this program cannot drift from it. See docs/design/measurement-result-parquet.md. - pip install jsonschema 'pyarrow>=19' rfc8785 # Python 3.11 or newer - python tools/json_to_parquet.py schemas/measurement-result/0.1.0/schema.json \ - results.parquet schemas/measurement-result/0.1.0/examples/*.json + pip install benchx + python tools/json_to_parquet.py results.parquet \ + src/benchx/schemas/measurement-result/0.1.0/examples/*.json """ -import hashlib -import json import sys -from datetime import datetime +from pathlib import Path -import jsonschema -import pyarrow as pa import pyarrow.parquet as pq -import rfc8785 -TIMESTAMP = pa.timestamp("us", tz="UTC") -SCALARS = {"string": pa.string(), "integer": pa.int64(), "number": pa.float64(), "boolean": pa.bool_()} +from benchx import core, parquet -def arrow_type(node, defs, json_type): - """Map one JSON Schema node to an Arrow type.""" - if "$ref" in node: - node = defs[node["$ref"].rsplit("/", 1)[-1]] - nested = lambda n: arrow_type(n, defs, json_type) # noqa: E731 - if "allOf" in node and "type" not in node: # a reference narrowed by assertions - return nested(node["allOf"][0]) - 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}") - required = set.intersection(*(set(b.get("required", [])) for b in objects)) - types = {} # in first-appearance order - for b in objects: - for k, v in b["properties"].items(): - t = nested(v) - if types.setdefault(k, t) != t: - raise ValueError(f"oneOf alternatives disagree on the type of {k!r}: {types[k]} and {t}") - return pa.struct([pa.field(k, t, k not in required) for k, t in types.items()]) - if "const" in node or "enum" in node: - literals = {type(v) for v in node.get("enum", [node.get("const")])} - if literals not in ({str}, {int}): # bool is not int here: type(True) is bool - raise ValueError(f"unsupported const or enum: {node}") - return pa.string() if literals == {str} else pa.int64() - if node.get("type") == "array": - return pa.list_(pa.field("element", nested(node["items"]), False)) - if node.get("type") == "object": - if node.get("additionalProperties") is not False or "properties" not in node: - return json_type - required = node.get("required", []) - return pa.struct([pa.field(k, nested(v), k not in required) for k, v in node["properties"].items()]) - if node.get("format") == "date-time": - return TIMESTAMP - scalar = node.get("type") - return SCALARS[scalar] if isinstance(scalar, str) else json_type # mixed primitives stay JSON - - -def to_row(data_type, value): - """Shape one message value for its Arrow type.""" - if value is None: - return None - if pa.types.is_struct(data_type): - if not isinstance(value, dict): # a bare observation number - value = {"value": value} - return {f.name: to_row(f.type, value.get(f.name)) for f in data_type} - if pa.types.is_list(data_type): - return [to_row(data_type.value_type, v) for v in value] - if data_type == pa.json_(): # open object as canonical JSON - return rfc8785.dumps(value).decode() - if data_type == TIMESTAMP: - return datetime.fromisoformat(value) - return value - - -def unique_keys(pairs): - # jsonschema only sees the parsed mapping, after duplicate keys are lost. - result = dict(pairs) - if len(result) != len(pairs): - raise ValueError(f"duplicate JSON key in {[key for key, _ in pairs]}") - return result - - -def non_finite(token): - # Python's JSON parser accepts NaN and Infinity even though JSON does not. - raise ValueError(f"invalid JSON number: {token}") - - -def load_json(path): - """Load strict I-JSON: UTF-8, unique keys, finite numbers, safe integers.""" - with open(path, encoding="utf-8") as f: - value = json.load(f, object_pairs_hook=unique_keys, parse_constant=non_finite) - rfc8785.dumps(value) # rejects numbers outside the RFC 8785 domain, such as integers beyond 2**53 - return value - - -def main(schema_path, output, *message_paths): - message_schema = load_json(schema_path) - messages = [load_json(path) for path in message_paths] - for message in messages: - jsonschema.validate(message, message_schema) - - defs = message_schema["$defs"] - # Build with strings, then cast: PyArrow cannot fill nested JSON fields - # from Python values directly. - build = arrow_type(message_schema, defs, pa.string()) - final = arrow_type(message_schema, defs, pa.json_()) - table = pa.Table.from_pylist([to_row(pa.struct(final), m) for m in messages], schema=pa.schema(build)) - table = table.cast(pa.schema(final)).replace_schema_metadata({ - "benchx.schema_uri": message_schema["$id"], - "benchx.message_schema_sha256": hashlib.sha256(rfc8785.dumps(message_schema)).hexdigest(), - }) - pq.write_table(table, output, compression="zstd") +def main(output, *paths): + """Write the documents at `paths` to `output`; nothing is written if one is invalid.""" + documents, failed = [], False + for path in paths: + inspection = core.inspect(Path(path), "measurement-result") + for error in inspection.errors: + print(f"{path}: {error}", file=sys.stderr) + failed |= not inspection.valid + documents.append(inspection.data) + if failed: + return 1 + pq.write_table(parquet.message_table(documents), output, compression="zstd") + return 0 if __name__ == "__main__": - main(*sys.argv[1:]) + if len(sys.argv) < 3: + sys.exit(f"usage: {sys.argv[0]} OUTPUT.parquet DOCUMENT.json...") + sys.exit(main(*sys.argv[1:])) diff --git a/zensical.toml b/zensical.toml index fe06bbc..2869700 100644 --- a/zensical.toml +++ b/zensical.toml @@ -39,6 +39,7 @@ nav = [ { "Comparator" = "design/comparator.md" }, { "Prototype scope" = "design/prototype-scope.md" }, { "Prototype design" = "design/prototype-design.md" }, + { "Result in Parquet" = "design/measurement-result-parquet.md" }, ] }, { "Schemas" = [ { "Overview" = "schemas/index.md" }, From 815bebcfbae94b3448889544e50bff6f0a910514 Mon Sep 17 00:00:00 2001 From: Rok Mihevc Date: Sun, 15 Nov 2026 01:11:31 +0100 Subject: [PATCH 6/6] BUG: reject oneOf shapes the Parquet mapping does not cover A oneOf was accepted whenever it had an object alternative. An open object was merged into the struct, dropping undeclared keys, and any non-object alternative was stored as if it were a bare observation number. Accept only closed objects, or one number beside one closed object with a number value. --- src/benchx/parquet.py | 11 +++++++++-- tests/test_parquet.py | 5 +++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/benchx/parquet.py b/src/benchx/parquet.py index c2b48da..6e643d6 100644 --- a/src/benchx/parquet.py +++ b/src/benchx/parquet.py @@ -37,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: diff --git a/tests/test_parquet.py b/tests/test_parquet.py index 221c463..c7dfc23 100644 --- a/tests/test_parquet.py +++ b/tests/test_parquet.py @@ -70,6 +70,11 @@ def closed(json_type): {"enum": ["a", 1]}, {"oneOf": [{"type": "string"}, {"type": "integer"}]}, {"oneOf": [closed("string"), closed("integer")]}, + {"oneOf": [closed("string"), {"type": "object", "properties": {"y": {"type": "string"}}}]}, # open + {"oneOf": [closed("string"), {"type": "object", "additionalProperties": False}]}, # no properties + {"oneOf": [{"type": "string"}, closed("string")]}, # a bare value that is not a number + {"oneOf": [{"type": "number"}, closed("number")]}, # no `value` to hold the bare number + {"oneOf": [{"type": "number"}, closed("number"), closed("number")]}, ]) def test_unsupported_nodes_fail(node): with pytest.raises(ValueError):