Stores & persistence

A store is where the BenchmarkRunner persists every BenchmarkRecord after execution — the third of qpubench’s three layers (schemas describe, adapters execute, stores persist). qpubench ships three stores. All implement the ResultStore protocol:

class ResultStore(Protocol):
    def save(self, record: BenchmarkRecord) -> None: ...
    def load(self, experiment_id: str) -> BenchmarkRecord: ...
    def query(self, **filters) -> list[BenchmarkRecord]: ...

NDJSONStore

Zero-dependency, append-only, one JSON record per line.
Suitable for streaming writes, grep-based ad-hoc queries, and compatibility with any NDJSON tool.

import pathlib
from qpubench import BenchmarkRunner, NDJSONStore

store  = NDJSONStore(pathlib.Path("results/sweep.ndjson"))
runner = BenchmarkRunner(store=store)

Querying

Keys use __ as a field-path separator:

# Filter by backend name
records = store.query(backend__name="aer_statevector")

# Multiple filters (AND)
records = store.query(result__status="succeeded", vqa__molecule="H2")

# Algorithm-driven runs
records = store.query(options__algorithm_spec__name="ADAPTVQE")

# Load a single record by UUID
record = store.load("3f2a1b4c-...")

# All records
all_records = store.all()

The query matches on the model_dump() representation of BenchmarkRecord. Nested dicts use the same __ convention, e.g. result__expectation_values would require a further traversal that the query engine does not yet support — load the record and filter in Python for deep nesting.

Thread safety

Individual write() calls on CPython are atomic for lines shorter than PIPE_BUF (~4 KB on Linux). For large records or concurrent writers, use an explicit file lock or separate per-worker files.

ParquetStore

Columnar store backed by a single Parquet file. Requires pip install 'qpubench[storage]'.

from qpubench.store import ParquetStore
import pathlib

store = ParquetStore(pathlib.Path("results/sweep.parquet"))

Records are flattened one level deep before writing. A _raw_json column stores the full JSON for lossless round-trips.

save() rewrites the whole file per record (Parquet files are immutable), so treat this as an export/analysis format: collect records during a sweep with NDJSONStore, then bulk-load them with store.save_many(records) (one rewrite total).

# Flat pandas DataFrame for analysis and plotting
df = store.to_dataframe()
print(df[["molecule", "final_eigenvalue", "circuit_depth"]].describe())

# Query — column-level only: filters use flat single-underscore column
# names (not NDJSONStore's dotted __ paths); unknown columns are ignored
records = store.query(backend_name="aer_statevector")
records = store.query(result_status="succeeded")

# Load by UUID (uses _raw_json → BenchmarkRecord round-trip)
record = store.load("3f2a1b4c-...")

Flat column names produced from a BenchmarkRecord:

Column	Source
`schema_version`, `experiment_id`, `run_id`, `timestamp`	Top-level
`num_qubits`, `circuit_depth`, `tags`, `notes`	Top-level
`circuit_computing_model`, `circuit_qubit_modality`, `circuit_format`	`circuit.*`
`backend_name`, `backend_provider`	`backend.*`
`shots`, `opt_level`, `error_mitigation`	`options.*`
`result_status`, `qpu_time_s`, `total_time_s`	`result.*`
`molecule`, `basis`, `ansatz`	`vqa.*` (inputs)
`final_eigenvalue`, `ground_truth`	`vqa_result.*` (computed outputs)
`_raw_json`	Full JSON (for `load()` round-trip)

S3Store

Object store backed by S3 or any S3-compatible endpoint. Requires pip install 'qpubench[s3]'.

One JSON object per record — f"{prefix}/{experiment_id}.json" — instead of NDJSONStore’s shared append-only file. There is no read-modify-write of shared state, so concurrent writers (separate processes, machines, or a distributed sweep) do not race. Only put_object, get_object, and the list_objects_v2 paginator are used, so it works unmodified against real AWS S3 and against S3-compatible gateways.

AWS S3

from qpubench import BenchmarkRunner, S3Store

store = S3Store(
    "my-results-bucket",
    prefix="sweeps/2026-07-06",
    region_name="eu-west-1",
    # aws_access_key_id / aws_secret_access_key are optional — omit them to
    # use the standard boto3 credential chain (env vars, ~/.aws/credentials,
    # instance role, etc.)
)
runner = BenchmarkRunner(store=store)

Any keyword accepted by boto3.client("s3", ...) can be passed through directly — endpoint_url, region_name, aws_access_key_id, aws_secret_access_key, config=botocore.config.Config(...) — which also makes S3Store work against MinIO or any other S3-compatible service. Pass an already-constructed client instead via client=:

import boto3

store = S3Store("my-bucket", client=boto3.client("s3", endpoint_url="http://localhost:9000"))

Hugging Face Storage Buckets

Storage Buckets expose an S3-compatible gateway at https://s3.hf.co/<namespace>. S3Store.huggingface() sets up the required botocore Config (path-style addressing, single region, ListObjectsV2-only) for you:

from qpubench import S3Store

store = S3Store.huggingface(
    bucket="my-training-bucket",
    namespace="my-username",          # or an organization name
    access_key_id="HFAK...",          # Settings → Access Tokens →
    secret_access_key="...",          #   (token) → Generate S3 credentials
    prefix="qpubench-results",
)
runner = BenchmarkRunner(store=store)

Generate the S3 credentials from a Hugging Face User Access Token with Write permission (its dropdown menu has a Generate S3 credentials action); the bucket itself must already exist (hf buckets create my-training-bucket, or via huggingface_hub.create_bucket).

Loading and querying

Identical API to NDJSONStore — save(), load(experiment_id), query(**filters), all():

record  = store.load("3f2a1b4c-...")
records = store.query(backend__name="aer_statevector", result__status="succeeded")
all_records = store.all()   # lists every object under the prefix, then fetches each

query()/all() list every object under the store’s prefix and fetch each in turn — fine for benchmark-scale result sets, but there is no server-side filtering, so very large buckets should be partitioned by prefix (e.g. one prefix per run_id) to keep listings small.

Hooks

Hooks run after every BenchmarkRecord is finalised but before it is saved to the store. Use them for live monitoring, intermediate logging, or side-channel metrics.

def log_energy(record):
    ev  = record.result.expectation_values
    val = f"{ev[0].value:.6f}" if ev else "n/a"
    print(f"[{record.backend.name}] {record.vqa.molecule or '?'}  E={val}  "
          f"t={record.result.total_time_s:.2f}s")

runner.add_hook(log_energy)

Multiple hooks are called in registration order. An exception inside a hook is logged but does not abort the run or prevent persistence.

# Hooks can also save intermediate state to a separate file
import json, pathlib

log_path = pathlib.Path("live_log.jsonl")

def side_log(record):
    with log_path.open("a") as f:
        f.write(json.dumps({
            "id":     record.experiment_id,
            "E":      record.result.expectation_values[0].value
                      if record.result.expectation_values else None,
            "status": record.result.status.value,
        }) + "\n")

runner.add_hook(side_log)

Sweep + store pattern

records = runner.sweep(
    circuits=[circuit_a, circuit_b],
    backend_names=["stub_gate", "aer_statevector"],
    options_list=[ExecutionOptions(shots=s) for s in [512, 2048, 8192]],
    run_id="h2_shots_sweep",
    tags=["vqe", "h2", "shots_sweep"],
)
# All 12 records (2×2×3) are saved automatically; group by run_id:
group = store.query(run_id="h2_shots_sweep")
print(f"{len(group)} records in sweep")