From ab44882c7134f9a0460fe1f9658765c41ffe8d3c Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Fri, 18 Sep 2026 17:06:59 -0400 Subject: [PATCH 1/3] docs(plans): ADR-009 digest stability (proposed) Proposes two changes so a heal is flagged only when the result changed: materialization runs single-partition, because a float SUM/AVG is not bit-reproducible under DataFusion's parallel partial aggregation, and result_digest becomes a digest of the snapshot's ordered Arrow content, computed from the file as read back, because file bytes move with the writer's version and with how the stream was batched. Amends ADR-004 and ADR-006 D5. scripts/spike_float_aggregate_digest.py: 6 distinct digests in 6 runs at default partitions, 1 at target_partitions=1, integer control stable. scripts/spike_logical_digest.py: the content digest is invariant to batch size, codec and row-group size, and detects a changed value, a null replaced by 0.0, and a row swap. Co-Authored-By: Claude Fable 5.1 --- plans/ADR-009-digest-stability.md | 227 +++++++++++++++++++++++ scripts/spike_float_aggregate_digest.py | 77 ++++++++ scripts/spike_logical_digest.py | 229 ++++++++++++++++++++++++ 3 files changed, 533 insertions(+) create mode 100644 plans/ADR-009-digest-stability.md create mode 100644 scripts/spike_float_aggregate_digest.py create mode 100644 scripts/spike_logical_digest.py diff --git a/plans/ADR-009-digest-stability.md b/plans/ADR-009-digest-stability.md new file mode 100644 index 0000000..696191f --- /dev/null +++ b/plans/ADR-009-digest-stability.md @@ -0,0 +1,227 @@ +# ADR: Digest stability (a heal is flagged only when the result changed) + +- **Status:** Proposed (2026-09-18). Amends + `plans/ADR-004-result-digest-canonical-ordering.md` (Option A's "hash the + snapshot bytes") and `plans/ADR-006-read-path-loads-builds.md` D5 + ("`result_digest` keeps its file-hash definition"). D5's canonical sort is + unchanged and is still required. +- **Context:** the 2026-09-18 cache audit (tallyman @ `a748ea6`, xorq 0.3.26, + xorq-datafusion 0.2.7, pyarrow 21.0.0). No ticket filed yet. +- **Affected code:** `src/tallyman_xorq/result_cache.py` + (`snapshot_file_digest`, `verify_result_faithful`, `_verify_self_heal`), + `src/tallyman_xorq/build.py` (the execute-once step), + `src/tallyman_xorq/backend.py` (a single-partition connection), + `src/tallyman_core/manifest.py` (`result_digest`), and the `materialize` + writer that `plans/ADR-007-tallyman-owned-materialization.md` D4 introduces. + D2 and D3 assume that writer. If ADR-007 were rejected, D1 would stand as + written and D2 would need restating against xorq's writer. +- **Related ADRs:** `plans/ADR-008-row-order-of-reads.md` (uses the same + single-partition setting for a different job, and has an open question this + digest would answer). +- **Evidence:** `scripts/spike_float_aggregate_digest.py`, + `scripts/spike_logical_digest.py`. + +## Problem + +`result_digest` is the SHA-256 of the snapshot file's bytes. The contract gives +it one job: witnessing that a later rematerialization reproduced the original. +When a heal's digest does not match, tallyman writes a durable +`unfaithful_heal` record, pushes an SSE event, wipes the entry's Buckaroo stat +cache, evicts its session, and pins and badges the entry (ADR-006 D7, D10, +D12). That response is right for a recipe that calls `sample()`. It is +expensive, sticky and misleading when nothing about the result changed, and +today two things trigger it without a change in the result. + +**1. A float aggregate is not bit-reproducible under parallel execution.** +DataFusion sums each partition separately and merges the partial sums in +arrival order. Float addition is not associative, so the low bits move from +run to run. A group-by over 3,000,000 rows with a float `SUM` and `AVG`, +canonically ordered, six runs per configuration: + +| Aggregate | Partitions | Distinct digests in 6 runs | Median | +| --- | --- | --- | --- | +| float | default (14) | 6 | 6 ms | +| float | `target_partitions = 1` | 1 | 20 ms | +| integer | default (14) | 1 | 7 ms | +| integer | `target_partitions = 1` | 1 | 31 ms | + +The largest relative difference between two float runs was 6.4e-16. At the +tallyman level the audit healed a float-aggregate entry six times: six +different files, six `unfaithful_heal` records, six stat-cache wipes. An +integer-only aggregate healed byte-identical six times. The canonical sort +cannot help, since the rows are in the same order and it is the values that +differ. Aggregate is the usual reason an entry is worthy, so most expensive +entries with a float measure are flagged after any eviction. The message +blames "execution (#83), a fixed graph that runs differently each execute", +which sends the author to a recipe that is deterministic. The default partition +count is also the machine's core count, so the same build merges differently +on a different machine. + +**2. File bytes depend on things that are not the result.** + +- How the stream was batched. The same rows handed to a parquet writer as + 100,000-row batches and as 8,192-row batches gave different files unless each + row group was first combined into contiguous arrays. +- The writer's version. The footer carries + `created_by = 'parquet-cpp-arrow version 21.0.0'`, so after a pyarrow upgrade + every heal would mismatch. +- Under xorq's writer today, whether the cache node was the root of the + executed expression (1,234 bytes against 858 for the same rows). + +ADR-004 accepted "a library upgrade changes the hash; rebuild is fine". That +was written before a mismatch became loud. With D7, D10 and D12 in place, an +upgrade would pin and badge every entry that gets evicted until the corpus is +rebuilt. + +## Decisions + +### D1. Materialization runs single-partition + +`materialize` (ADR-007 D4) executes on a connection configured with +`SET datafusion.execution.target_partitions = 1`. The build and every heal use +it, so both run one plan with one merge order, on any machine. It is a +different connection object from ADR-008's window connection, with the same +setting, because a long bake must not share a context with page reads. + +Cost: about 3x on the spike's aggregate. ADR-004 measured a 3.1M-group +aggregate at 0.5 s parallel against 3.4 to 3.9 s single-partition, and a full +43-column read at 2.5 s against 6.9 s, on the 11.8M-row parking file. A bake +runs once per entry and once per heal, never on a read. + +ADR-004 rejected pinning scan order through session config as the *ingest* +lever, because it serializes everything downstream. This decision uses the same +setting for a different job and accepts that cost knowingly: reproducible +arithmetic is the point, and no setting gives both. + +**Gate.** Rebuild the parking corpus under both settings before committing. If +the single-partition rebuild is unacceptable, fall back to **D1b**: only plans +with a floating-point reduction or window run single-partition. That choice is +made once, at build, from the expression, and recorded in the manifest, so a +heal never re-derives it. + +*Rejected:* round floats before hashing. A value next to a rounding boundary +flips under one unit of noise in the last place, and among millions of values +some are. +*Rejected:* compare with a tolerance. When a heal runs the original values are +gone and only the digest is left. +*Rejected:* treat a mismatch as advisory when the plan has a float aggregate. +That removes verification from the entries most likely to be expensive. +*Rejected:* drop the canonical sort, since single-partition output is ordered +anyway. The sort is what puts the author's keys first in stored order +(ADR-008 D1 relies on it), and it keeps the result independent of the hash +table's emission order, which an engine upgrade can change. + +### D2. `result_digest` is a digest of the snapshot's content, computed from the file as read back + +The digest is a SHA-256 over the snapshot's ordered Arrow data: + +- one hash stream per column for validity (one byte per row), one for lengths + (variable-width types) and one for values, with null slots zeroed or emptied, + since a null slot may hold anything; +- each stream seeded with the column name and its logical type, where `string`, + `large_string` and `string_view` are one type; +- the streams combined in schema order together with the row count; +- stored with an algorithm prefix (`arrow-sha256:`) so a future definition can + never be compared against this one by accident. + +Separate streams matter. The first version of the spike fed validity and +values into one hasher per column, and the digest then depended on where batch +boundaries fell. + +One function computes it, from the written file read back, for both the build +and verify. This is the "one derivation route" rule behind ADR-006 D8. The +spike shows why it cannot be computed from the stream handed to the writer: the +writer coerced a `timestamp[s]` column to `timestamp[ms]`, and the two digests +differed. + +Measured on 3,000,000 rows with nulls, NaNs, strings, booleans and timestamps: + +| Check | Result | +| --- | --- | +| Same rows as 8,192-row, 100,000-row and single batches | 1 digest | +| Same file read back in 1,000-row batches | same | +| Same rows written Snappy with 8,192-row groups instead of zstd with 1,048,576 | same | +| Same file read through DataFusion with `order_by(id)` | same | +| One value changed | differs | +| One null replaced by `0.0` | differs | +| First two rows swapped | differs | +| Read back and hash a 57 MB file (127 MB of Arrow data) | 0.12 s, against 0.02 s for a file-bytes hash | + +The digest stays order-sensitive, so D5's canonical sort is still what makes it +reproducible. + +This reverses part of ADR-006 D5's reasoning. D5 rejected the multiset digest +partly because "verify must read every row and the digest stops being a hash +of the artifact". Both are true of this digest as well. They are accepted here +because verify runs only on a heal and in the sweep, at roughly 1 GB/s of Arrow +data, and because being a hash of the artifact is exactly what turns a writer +upgrade into a false alarm. + +*Rejected:* keep file bytes and pin the writer's settings. It is reproducible +today (the spike gets one digest across batch sizes once each row group is +combined) and six times cheaper to verify. It also freezes the codec and +row-group size for the life of the corpus, and it fails on the first pyarrow +upgrade. +*Rejected:* polars `hash_rows`. Its documentation does not guarantee stable +results across polars versions. + +### D3. The snapshot's format + +Tallyman's writer (ADR-007 D4) regroups the record-batch stream into row groups +of 1,048,576 rows, combines each row group into contiguous arrays, and writes +zstd level 3, parquet format 2.6, statistics on. For the spike's rows that is +57.1 MB, 3 row groups and a 2.7 KB footer, against 100.3 MB, 367 row groups +and a 228.5 KB footer in the shape xorq writes (one Snappy row group per +8,192-row batch). Memory is bounded by one row group. + +Combining each row group keeps the file bytes reproducible as well. Nothing +depends on that after D2, and it means two bakes can still be compared with +`cmp` when debugging. Because of D2 these settings can change later without +touching any digest. + +### D4. A mismatch record names its likely cause + +With D1 and D2 in place the causes left are the recipe's own nondeterminism +(`sample()`, `now()`, an impure UDF), source drift under `off` identity mode, +and an engine upgrade that changed results. The manifest records the xorq, +xorq-datafusion and pyarrow versions at build, and the `unfaithful_heal` record +carries them alongside the versions at heal. When they differ, the message says +so instead of blaming the recipe. The contract's attribution table gains that +row. + +### D5. It lands with the rebuild + +Every worthy entry's digest is recomputed by the corpus rebuild of ADR-007 D9. +The manifest field keeps its name. Tests that cannot fail first ride with the +fix; the float-aggregate heal test and a batch-boundary digest test fail on +`main` and belong in the failing-tests commit. + +## Consequences + +- A float-aggregate entry heals to the same digest, and a pyarrow upgrade no + longer flags anything. The D10 and D12 responses are left for the causes they + were designed for. +- Bakes and heals are slower, by about 3x on aggregation at spike scale and up + to 7x in ADR-004's parking measurement. Reads are unaffected. +- Verify decodes the file instead of hashing its bytes. It runs on a heal and + in `catalog_scan_staleness(verify_results=True)`, never on a read. +- Snapshots are smaller, and their footers were 85 times smaller in the spike, + which matters to every page request that opens one. +- `docs/system-contract.md` changes in "Result digest" and in the manifest + table ("SHA-256 of the baked result snapshot"), and its verification table + gains the engine-change row. +- The same function can digest the CSV intermediate (ADR-008, open + question 1). + +## Open questions + +1. **Rebuild time under single-partition bakes.** Decides D1 against D1b. +2. **Nested types.** The spike covers fixed-width, boolean, string and binary + columns. Lists, structs and maps need a recursive definition. +3. **Types parquet cannot store as given.** A `timestamp[s]` column comes back + as `timestamp[ms]`, so the snapshot's schema differs from the entry's + recorded schema. Either the writer refuses such a column, or the entry's + schema is recorded from the snapshot. +4. **Engine upgrades.** Single-partition execution fixes the merge order within + one engine version. Nothing guarantees float results across versions. D4 + attributes that case and the remedy stays a rebuild. diff --git a/scripts/spike_float_aggregate_digest.py b/scripts/spike_float_aggregate_digest.py new file mode 100644 index 0000000..55f2c67 --- /dev/null +++ b/scripts/spike_float_aggregate_digest.py @@ -0,0 +1,77 @@ +"""ADR-009 evidence: a float aggregate is not bit-reproducible under DataFusion's parallel partial aggregation. + +Runs the same canonically ordered group-by six times per configuration and hashes the result's Arrow buffers. +A float ``SUM``/``AVG`` comes back with different low bits from run to run under the default partition count, and +with identical bits under ``target_partitions = 1``. An integer-only aggregate is the control: it is identical in +both configurations. Median wall time is printed so the cost of single-partition execution is visible. + + uv run python scripts/spike_float_aggregate_digest.py +""" + +from __future__ import annotations + +import hashlib +import statistics +import tempfile +import time +from pathlib import Path + +import numpy as np +import pyarrow as pa +import pyarrow.parquet as pq +import xorq.api as xo + +N = 3_000_000 +RUNS = 6 + + +def value_digest(table: pa.Table) -> str: + h = hashlib.sha256() + for col in table.columns: + for chunk in col.chunks: + for buf in chunk.buffers(): + if buf is not None: + h.update(buf) + return h.hexdigest()[:12] + + +def run(path: Path, partitions: int | None, kind: str) -> tuple[str, float, pa.Table]: + con = xo.connect() + if partitions is not None: + con.raw_sql(f"SET datafusion.execution.target_partitions = {partitions}") + t = con.read_parquet(str(path)) + if kind == "float": + expr = t.group_by("g").agg(n=t.count(), s=t.v.sum(), m=t.v.mean()).order_by("g") + else: + expr = t.group_by("g").agg(n=t.count(), s=t.k.sum(), hi=t.k.max()).order_by("g") + t0 = time.perf_counter() + table = expr.to_pyarrow() + return value_digest(table), time.perf_counter() - t0, table + + +def main() -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "t.parquet" + rng = np.random.default_rng(5) + source = pa.table( + {"g": rng.integers(0, 1_000, N), "v": rng.random(N) * 1e6, "k": rng.integers(0, 1_000_000, N)} + ) + pq.write_table(source, path, row_group_size=100_000) + groups = pq.ParquetFile(path).metadata.num_row_groups + print(f"source: {N:,} rows, {groups} row groups, {path.stat().st_size / 1e6:.1f} MB\n") + for kind in ("float", "int"): + for label, partitions in (("default partitions", None), ("target_partitions=1", 1)): + results = [run(path, partitions, kind) for _ in range(RUNS)] + digests = [d for d, _, _ in results] + median_ms = statistics.median(s for _, s, _ in results) * 1000 + distinct = len(set(digests)) + line = f"{kind:5s} {label:20s} distinct digests over {RUNS} runs: {distinct}, median {median_ms:.0f} ms" + if kind == "float" and len(set(digests)) > 1: + a, b = results[0][2].column("s").to_numpy(), results[1][2].column("s").to_numpy() + rel = np.max(np.abs(a - b) / np.abs(a)) + line += f", max relative difference between two runs {rel:.1e}" + print(line) + + +if __name__ == "__main__": + main() diff --git a/scripts/spike_logical_digest.py b/scripts/spike_logical_digest.py new file mode 100644 index 0000000..7a695d1 --- /dev/null +++ b/scripts/spike_logical_digest.py @@ -0,0 +1,229 @@ +"""ADR-009 evidence: two candidate definitions of ``result_digest`` for a tallyman-written snapshot. + +**File bytes** (today's definition). Reproducible only if the writer's input and settings are pinned: the same rows +arriving in differently sized record batches must be regrouped into fixed row groups, and each row group combined +into contiguous arrays before it is written. The parquet footer also embeds the writer's version string, so the +bytes move on every pyarrow upgrade. + +**Logical content**. A SHA-256 over the ordered Arrow data, one hasher per column, fed a normalized form that does +not depend on batch boundaries, on the physical string type, or on whatever sits in null slots. It is independent +of codec, row-group size and writer version, and costs a read-back to verify. + +The script checks both for the invariances each needs, compares the file shape of a tallyman-side writer with +the shape xorq's ``ParquetStorage`` writes, then times the two digests. + + uv run python scripts/spike_logical_digest.py +""" + +from __future__ import annotations + +import hashlib +import tempfile +import time +from pathlib import Path + +import numpy as np +import pyarrow as pa +import pyarrow.compute as pc +import pyarrow.parquet as pq +import xorq.api as xo + +N = 3_000_000 +ROW_GROUP = 1_048_576 + + +def _logical_type(dtype: pa.DataType) -> str: + if pa.types.is_string(dtype) or pa.types.is_large_string(dtype) or pa.types.is_string_view(dtype): + return "string" + if pa.types.is_binary(dtype) or pa.types.is_large_binary(dtype) or pa.types.is_binary_view(dtype): + return "binary" + return str(dtype) + + +class LogicalDigest: + """Order-sensitive digest of a record-batch stream, invariant to chunking and physical encoding.""" + + # Each column keeps one hasher per byte stream (validity, lengths, values). A single hasher per column would + # interleave the streams chunk by chunk, and the digest would then depend on where the batch boundaries fall. + STREAMS = ("validity", "lengths", "values") + + def __init__(self, schema: pa.Schema): + self.schema = schema + self.hashers = [ + {s: hashlib.sha256(f"{f.name}\x00{_logical_type(f.type)}\x00{s}".encode()) for s in self.STREAMS} + for f in schema + ] + self.rows = 0 + + def update(self, batch: pa.RecordBatch | pa.Table) -> None: + for hashers, column in zip(self.hashers, batch.columns): + chunks = column.chunks if isinstance(column, pa.ChunkedArray) else [column] + for arr in chunks: + self._update_array(hashers, arr) + self.rows += batch.num_rows + + @staticmethod + def _update_array(hashers: dict, arr: pa.Array) -> None: + if pa.types.is_dictionary(arr.type): + arr = arr.dictionary_decode() + nulls = arr.is_null().to_numpy(zero_copy_only=False) + hashers["validity"].update(nulls.astype(np.uint8).tobytes()) # one byte per row, null or not + dtype = arr.type + if _logical_type(dtype) in ("string", "binary"): + target = pa.large_string() if _logical_type(dtype) == "string" else pa.large_binary() + filled = pc.fill_null(arr.cast(target), "" if target == pa.large_string() else b"") + hashers["lengths"].update( + pc.binary_length(filled).to_numpy(zero_copy_only=False).astype(np.int64).tobytes() + ) + offsets = np.frombuffer(filled.buffers()[1], dtype=np.int64)[ + filled.offset : filled.offset + len(filled) + 1 + ] + if len(filled): + hashers["values"].update(memoryview(filled.buffers()[2])[int(offsets[0]) : int(offsets[-1])]) + elif pa.types.is_boolean(dtype): + hashers["values"].update(pc.fill_null(arr, False).to_numpy(zero_copy_only=False).astype(np.uint8).tobytes()) + elif pa.types.is_primitive(dtype) or pa.types.is_decimal(dtype) or pa.types.is_fixed_size_binary(dtype): + width = dtype.bit_width // 8 + raw = np.frombuffer(arr.buffers()[1], dtype=np.uint8)[arr.offset * width : (arr.offset + len(arr)) * width] + if arr.null_count: + raw = raw.reshape(len(arr), width).copy() + raw[nulls] = 0 # null slots may hold anything; zero them + hashers["values"].update(raw.tobytes() if arr.null_count else memoryview(raw)) + else: + raise NotImplementedError(f"nested/other type not covered by the spike: {dtype}") + + def hexdigest(self) -> str: + top = hashlib.sha256(str(self.rows).encode()) + for hashers in self.hashers: + for stream in self.STREAMS: + top.update(hashers[stream].digest()) + return top.hexdigest()[:16] + + +def logical_digest(batches, schema: pa.Schema) -> str: + d = LogicalDigest(schema) + for batch in batches: + d.update(batch) + return d.hexdigest() + + +def write_pinned(batches, schema: pa.Schema, dest: Path, combine: bool = True, **settings) -> str: + """Regroup a batch stream into fixed row groups and write it; return the digest of the file's bytes.""" + options = {"compression": "zstd", "compression_level": 3, "version": "2.6", "data_page_version": "1.0"} | settings + with pq.ParquetWriter(dest, schema, **options) as writer: + pending, rows = [], 0 + for batch in batches: + pending.append(batch) + rows += batch.num_rows + while rows >= ROW_GROUP: + table = pa.Table.from_batches(pending) + head, tail = table.slice(0, ROW_GROUP), table.slice(ROW_GROUP) + writer.write_table(head.combine_chunks() if combine else head, row_group_size=ROW_GROUP) + pending, rows = tail.to_batches(), tail.num_rows + if rows: + table = pa.Table.from_batches(pending) + writer.write_table(table.combine_chunks() if combine else table, row_group_size=ROW_GROUP) + return hashlib.sha256(dest.read_bytes()).hexdigest()[:16] + + +def make_table() -> pa.Table: + rng = np.random.default_rng(11) + floats = rng.random(N) + floats[rng.integers(0, N, 1000)] = np.nan + ints = rng.integers(0, 1_000_000, N) + return pa.table( + { + "id": pa.array(np.arange(N)), + "g": pa.array(ints), + "f": pa.array(floats, mask=rng.random(N) < 0.01), + "s": pa.array([f"k{v % 5000:05d}" for v in ints], mask=rng.random(N) < 0.01), + "b": pa.array(ints % 2 == 0), + "ts": pa.array(ints.astype("datetime64[s]")), + } + ) + + +def main() -> None: + table = make_table() + schema = table.schema + print(f"table: {N:,} rows, {table.nbytes / 1e6:.0f} MB of Arrow data, pyarrow {pa.__version__}\n") + with tempfile.TemporaryDirectory() as tmp: + tmp = Path(tmp) + + print("file bytes, the same rows arriving in different batch sizes:") + for combine in (True, False): + digests = { + size: write_pinned( + table.to_batches(max_chunksize=size), schema, tmp / f"c{int(combine)}_{size}.parquet", combine + ) + for size in (8_192, 100_000, N) + } + distinct = sorted(set(digests.values())) + print(f" combine_chunks per row group={combine!s:5}: {len(distinct)} distinct digest(s) {distinct}") + created_by = pq.ParquetFile(tmp / "c1_8192.parquet").metadata.created_by + print(f" footer created_by = {created_by!r} (moves with every writer upgrade)\n") + + print("file shape, the same rows:") + xorq_shape = tmp / "xorq_shape.parquet" + with pq.ParquetWriter(xorq_shape, schema, compression="snappy") as writer: + for batch in table.to_batches(max_chunksize=8_192): + writer.write_batch(batch) # one row group per batch, as xorq's ParquetStorage writes them + for label, path in (("zstd, 1,048,576-row groups", tmp / "c1_8192.parquet"), ("xorq's shape", xorq_shape)): + md = pq.ParquetFile(path).metadata + size = path.stat().st_size / 1e6 + groups, footer = md.num_row_groups, md.serialized_size / 1e3 + print(f" {label:27s}: {size:5.1f} MB, {groups:3d} row groups, {footer:6.1f} KB footer") + print() + + def of_file(path: Path, batch_size: int = 65_536) -> str: + pf = pq.ParquetFile(path) + return logical_digest(pf.iter_batches(batch_size=batch_size), pf.schema_arrow) + + print("logical content:") + by_batch = {size: logical_digest(table.to_batches(max_chunksize=size), schema) for size in (8_192, 100_000, N)} + distinct = len(set(by_batch.values())) + print(f" in-memory stream, batch sizes 8,192 / 100,000 / one table: {distinct} distinct digest(s)") + + zstd_path, snappy_path = tmp / "c1_8192.parquet", tmp / "snappy_small_groups.parquet" + pq.write_table(table, snappy_path, compression="snappy", row_group_size=8_192) + reference = of_file(zstd_path) + stored = pq.ParquetFile(zstd_path).schema_arrow + coerced = [f"{a.name}: {a.type} -> {b.type}" for a, b in zip(schema, stored) if a.type != b.type] + print(f" stream handed to the writer vs the file read back: same={by_batch[8_192] == reference}") + print(f" (the writer coerced {coerced})") + print(f" same file, read in 1,000-row batches: same={of_file(zstd_path, 1_000) == reference}") + print(f" same rows written snappy with 8,192-row groups: same={of_file(snappy_path) == reference}") + engine = xo.connect().read_parquet(str(zstd_path)).order_by("id").to_pyarrow_batches() + print(f" same file through DataFusion (read, order_by id): same={logical_digest(engine, stored) == reference}") + + def rewritten(changed: pa.Table) -> str: + path = tmp / "changed.parquet" + pq.write_table(changed, path, compression="zstd", row_group_size=ROW_GROUP) + return of_file(path) + + flipped = table.set_column(1, "g", pc.add(table["g"], pa.array(np.eye(1, N, 12345, dtype=np.int64)[0]))) + print(f" one value changed: differs={rewritten(flipped) != reference}") + f = table["f"].combine_chunks() + first_null = int(np.flatnonzero(f.is_null().to_numpy(zero_copy_only=False))[0]) + as_zero = pa.concat_arrays([f.slice(0, first_null), pa.array([0.0]), f.slice(first_null + 1)]) + print(f" one null replaced by 0.0: differs={rewritten(table.set_column(2, 'f', as_zero)) != reference}") + swapped = table.take(pa.array(np.r_[1, 0, np.arange(2, N)])) + print(f" first two rows swapped: differs={rewritten(swapped) != reference}\n") + + print("cost:") + t0 = time.perf_counter() + logical_digest(table.to_batches(max_chunksize=8_192), schema) + secs = time.perf_counter() - t0 + print(f" logical digest of an in-memory stream: {secs:.2f} s ({table.nbytes / 1e6 / secs:.0f} MB/s)") + t0 = time.perf_counter() + of_file(zstd_path) + secs = time.perf_counter() - t0 + print(f" logical digest of the file (read back + hash; the build and the verify path): {secs:.2f} s") + t0 = time.perf_counter() + hashlib.sha256(zstd_path.read_bytes()).hexdigest() + secs = time.perf_counter() - t0 + print(f" file-bytes digest to verify ({zstd_path.stat().st_size / 1e6:.0f} MB file): {secs:.2f} s") + + +if __name__ == "__main__": + main() From a3a04d5790f2faf87ab9af9406280a7091220b18 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sun, 20 Sep 2026 09:53:31 -0400 Subject: [PATCH 2/3] docs(plans): ADR-009 defines its terms and takes two format requirements from ADR-008 Adds a Terms section, writes every other ADR's decision label with its ADR number and what it decides, and replaces "bake" with "materialize". D3 (the snapshot's format) gains two requirements from the revised ADR-008: the writer numbers rows in a last column named __row_order, and it writes a parquet page index, which takes a range request from 79-90 ms to 19-24 ms at any depth. The digest covers __row_order like any other column. Co-Authored-By: Claude Fable 5.1 --- plans/ADR-009-digest-stability.md | 98 ++++++++++++++++++++++--------- 1 file changed, 69 insertions(+), 29 deletions(-) diff --git a/plans/ADR-009-digest-stability.md b/plans/ADR-009-digest-stability.md index 696191f..9855d36 100644 --- a/plans/ADR-009-digest-stability.md +++ b/plans/ADR-009-digest-stability.md @@ -1,10 +1,15 @@ # ADR: Digest stability (a heal is flagged only when the result changed) -- **Status:** Proposed (2026-09-18). Amends +- **Status:** Proposed (2026-09-18, revised 2026-09-20: D3 gains two format + requirements from `plans/ADR-008-row-order-of-reads.md`). Amends `plans/ADR-004-result-digest-canonical-ordering.md` (Option A's "hash the - snapshot bytes") and `plans/ADR-006-read-path-loads-builds.md` D5 - ("`result_digest` keeps its file-hash definition"). D5's canonical sort is - unchanged and is still required. + snapshot bytes") and decision D5 of + `plans/ADR-006-read-path-loads-builds.md` (the canonical sort), which said + "`result_digest` keeps its file-hash definition". The canonical sort itself + is unchanged and is still required. +- **Reading decision labels:** a bare label such as "D2" in this document + always means this ADR's own decision. Another ADR's decision is always + written with its ADR number and a few words saying what it decides. - **Context:** the 2026-09-18 cache audit (tallyman @ `a748ea6`, xorq 0.3.26, xorq-datafusion 0.2.7, pyarrow 21.0.0). No ticket filed yet. - **Affected code:** `src/tallyman_xorq/result_cache.py` @@ -12,7 +17,8 @@ `src/tallyman_xorq/build.py` (the execute-once step), `src/tallyman_xorq/backend.py` (a single-partition connection), `src/tallyman_core/manifest.py` (`result_digest`), and the `materialize` - writer that `plans/ADR-007-tallyman-owned-materialization.md` D4 introduces. + writer that decision D4 of `plans/ADR-007-tallyman-owned-materialization.md` + introduces (one writer for snapshots, used by the build and by every heal). D2 and D3 assume that writer. If ADR-007 were rejected, D1 would stand as written and D2 would need restating against xorq's writer. - **Related ADRs:** `plans/ADR-008-row-order-of-reads.md` (uses the same @@ -21,14 +27,32 @@ - **Evidence:** `scripts/spike_float_aggregate_digest.py`, `scripts/spike_logical_digest.py`. +## Terms + +- **Materialize:** run an entry's computation once and write the result to a + parquet file. That file is the entry's **snapshot**. +- **Worthy entry:** an entry tallyman materializes. A cheap entry is one it + does not, and it records no digest. +- **`result_digest`:** the value recorded in a worthy entry's manifest when its + snapshot is first written, against which every later rewrite is checked. +- **Heal:** re-create a snapshot that is missing from disk by re-running the + entry's build. +- **Unfaithful heal:** a heal whose digest does not match the recorded one. +- **Canonical sort:** the fixed total order in which a snapshot's rows are + written (the author's `order_by` keys, then an inherited `__row_order`, then + the remaining columns), from ADR-006 decision D5. +- **Partition:** one of the parallel streams DataFusion splits a query into. + `target_partitions = 1` runs a query as a single stream. + ## Problem `result_digest` is the SHA-256 of the snapshot file's bytes. The contract gives it one job: witnessing that a later rematerialization reproduced the original. When a heal's digest does not match, tallyman writes a durable `unfaithful_heal` record, pushes an SSE event, wipes the entry's Buckaroo stat -cache, evicts its session, and pins and badges the entry (ADR-006 D7, D10, -D12). That response is right for a recipe that calls `sample()`. It is +cache, evicts its session, and pins and badges the entry (ADR-006 decisions D7, +loud verification; D10, the Buckaroo-state wipe; and D12, the pin and badge). +That response is right for a recipe that calls `sample()`. It is expensive, sticky and misleading when nothing about the result changed, and today two things trigger it without a change in the result. @@ -69,7 +93,8 @@ on a different machine. executed expression (1,234 bytes against 858 for the same rows). ADR-004 accepted "a library upgrade changes the hash; rebuild is fine". That -was written before a mismatch became loud. With D7, D10 and D12 in place, an +was written before a mismatch became loud. With those three ADR-006 decisions +in place, an upgrade would pin and badge every entry that gets evicted until the corpus is rebuilt. @@ -77,16 +102,18 @@ rebuilt. ### D1. Materialization runs single-partition -`materialize` (ADR-007 D4) executes on a connection configured with +`materialize` (ADR-007 decision D4, the one writer of snapshots) executes on a +connection configured with `SET datafusion.execution.target_partitions = 1`. The build and every heal use it, so both run one plan with one merge order, on any machine. It is a different connection object from ADR-008's window connection, with the same -setting, because a long bake must not share a context with page reads. +setting, because a long materialization must not share a context with page +reads. Cost: about 3x on the spike's aggregate. ADR-004 measured a 3.1M-group aggregate at 0.5 s parallel against 3.4 to 3.9 s single-partition, and a full -43-column read at 2.5 s against 6.9 s, on the 11.8M-row parking file. A bake -runs once per entry and once per heal, never on a read. +43-column read at 2.5 s against 6.9 s, on the 11.8M-row parking file. A +materialization runs once per entry and once per heal, never on a read. ADR-004 rejected pinning scan order through session config as the *ingest* lever, because it serializes everything downstream. This decision uses the same @@ -107,8 +134,8 @@ gone and only the digest is left. *Rejected:* treat a mismatch as advisory when the plan has a float aggregate. That removes verification from the entries most likely to be expensive. *Rejected:* drop the canonical sort, since single-partition output is ordered -anyway. The sort is what puts the author's keys first in stored order -(ADR-008 D1 relies on it), and it keeps the result independent of the hash +anyway. The sort is what puts the author's keys first in the order the rows +are numbered (`__row_order`, ADR-008 decision D2), and it keeps the result independent of the hash table's emission order, which an engine upgrade can change. ### D2. `result_digest` is a digest of the snapshot's content, computed from the file as read back @@ -129,7 +156,8 @@ values into one hasher per column, and the digest then depended on where batch boundaries fell. One function computes it, from the written file read back, for both the build -and verify. This is the "one derivation route" rule behind ADR-006 D8. The +and verify. This is the "one derivation route" rule behind ADR-006 decision D8 +(the snapshot-key tripwire: the build and the read share one derivation). The spike shows why it cannot be computed from the stream handed to the writer: the writer coerced a `timestamp[s]` column to `timestamp[ms]`, and the two digests differed. @@ -147,10 +175,12 @@ Measured on 3,000,000 rows with nulls, NaNs, strings, booleans and timestamps: | First two rows swapped | differs | | Read back and hash a 57 MB file (127 MB of Arrow data) | 0.12 s, against 0.02 s for a file-bytes hash | -The digest stays order-sensitive, so D5's canonical sort is still what makes it -reproducible. +The digest stays order-sensitive, so the canonical sort is still what makes it +reproducible. `__row_order` is a column of the file like any other, so the +digest covers it. -This reverses part of ADR-006 D5's reasoning. D5 rejected the multiset digest +This reverses part of the reasoning in ADR-006 decision D5 (the canonical +sort). That decision rejected the multiset digest partly because "verify must read every row and the digest stops being a hash of the artifact". Both are true of this digest as well. They are accepted here because verify runs only on a heal and in the sweep, at roughly 1 GB/s of Arrow @@ -167,16 +197,23 @@ results across polars versions. ### D3. The snapshot's format -Tallyman's writer (ADR-007 D4) regroups the record-batch stream into row groups -of 1,048,576 rows, combines each row group into contiguous arrays, and writes -zstd level 3, parquet format 2.6, statistics on. For the spike's rows that is +Tallyman's writer (ADR-007 decision D4) regroups the record-batch stream into +row groups of 1,048,576 rows, combines each row group into contiguous arrays, +and writes zstd level 3, parquet format 2.6, statistics on. For the spike's rows that is 57.1 MB, 3 row groups and a 2.7 KB footer, against 100.3 MB, 367 row groups and a 228.5 KB footer in the shape xorq writes (one Snappy row group per 8,192-row batch). Memory is bounded by one row group. +ADR-008 adds two requirements. The writer numbers the rows in a last column +named `__row_order` (ADR-008 decision D2). And it writes a parquet page index, +which is what lets a page be fetched as a range of `__row_order` values without +decoding a whole row group: 19 to 24 ms at any depth with the index against 79 +to 90 ms without it, on a 287 MB file (`scripts/spike_row_order_paging.py` on +the ADR-008 branch). + Combining each row group keeps the file bytes reproducible as well. Nothing -depends on that after D2, and it means two bakes can still be compared with -`cmp` when debugging. Because of D2 these settings can change later without +depends on that after D2, and it means two writes of the same entry can still +be compared with `cmp` when debugging. Because of D2 these settings can change later without touching any digest. ### D4. A mismatch record names its likely cause @@ -191,7 +228,8 @@ row. ### D5. It lands with the rebuild -Every worthy entry's digest is recomputed by the corpus rebuild of ADR-007 D9. +Every worthy entry's digest is recomputed by the corpus rebuild of ADR-007 +decision D9 (one change, one rebuild). The manifest field keeps its name. Tests that cannot fail first ride with the fix; the float-aggregate heal test and a batch-boundary digest test fail on `main` and belong in the failing-tests commit. @@ -199,9 +237,9 @@ fix; the float-aggregate heal test and a batch-boundary digest test fail on ## Consequences - A float-aggregate entry heals to the same digest, and a pyarrow upgrade no - longer flags anything. The D10 and D12 responses are left for the causes they - were designed for. -- Bakes and heals are slower, by about 3x on aggregation at spike scale and up + longer flags anything. The loud responses of ADR-006 decisions D10 and D12 + are left for the causes they were designed for. +- Materializations and heals are slower, by about 3x on aggregation at spike scale and up to 7x in ADR-004's parking measurement. Reads are unaffected. - Verify decodes the file instead of hashing its bytes. It runs on a heal and in `catalog_scan_staleness(verify_results=True)`, never on a read. @@ -215,13 +253,15 @@ fix; the float-aggregate heal test and a batch-boundary digest test fail on ## Open questions -1. **Rebuild time under single-partition bakes.** Decides D1 against D1b. +1. **Rebuild time under single-partition materialization.** Decides D1 against + D1b. 2. **Nested types.** The spike covers fixed-width, boolean, string and binary columns. Lists, structs and maps need a recursive definition. 3. **Types parquet cannot store as given.** A `timestamp[s]` column comes back as `timestamp[ms]`, so the snapshot's schema differs from the entry's recorded schema. Either the writer refuses such a column, or the entry's - schema is recorded from the snapshot. + schema is recorded from the snapshot. `__row_order` pushes toward the second + answer, since the writer adds a column the entry's graph does not have. 4. **Engine upgrades.** Single-partition execution fixes the merge order within one engine version. Nothing guarantees float results across versions. D4 attributes that case and the remedy stays a rebuild. From 95df5f7127e9b1230b97a7f311326b09dd76d3c2 Mon Sep 17 00:00:00 2001 From: Paddy Mullen Date: Sun, 20 Sep 2026 10:16:50 -0400 Subject: [PATCH 3/3] docs(plans): ADR-009 D1 drops its speed gate Paddy, in the grilling session: a cohesive system that works reliably comes first, and speed problems are handled as they come up. Every materialization runs single-partition. The float-only variant is noted as a later option and is no longer a gate on the decision. Co-Authored-By: Claude Fable 5.1 --- plans/ADR-009-digest-stability.md | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/plans/ADR-009-digest-stability.md b/plans/ADR-009-digest-stability.md index 9855d36..5f9763c 100644 --- a/plans/ADR-009-digest-stability.md +++ b/plans/ADR-009-digest-stability.md @@ -1,7 +1,8 @@ # ADR: Digest stability (a heal is flagged only when the result changed) - **Status:** Proposed (2026-09-18, revised 2026-09-20: D3 gains two format - requirements from `plans/ADR-008-row-order-of-reads.md`). Amends + requirements from `plans/ADR-008-row-order-of-reads.md`, and D1 lost its + speed gate). Amends `plans/ADR-004-result-digest-canonical-ordering.md` (Option A's "hash the snapshot bytes") and decision D5 of `plans/ADR-006-read-path-loads-builds.md` (the canonical sort), which said @@ -120,11 +121,13 @@ lever, because it serializes everything downstream. This decision uses the same setting for a different job and accepts that cost knowingly: reproducible arithmetic is the point, and no setting gives both. -**Gate.** Rebuild the parking corpus under both settings before committing. If -the single-partition rebuild is unacceptable, fall back to **D1b**: only plans -with a floating-point reduction or window run single-partition. That choice is -made once, at build, from the expression, and recorded in the manifest, so a -heal never re-derives it. +Paddy's call in the grilling session (2026-09-20): "I want a cohesive system +that works reliably, then we can worry about speed problems as they come up." +So every materialization runs single-partition, and this decision carries no +speed gate. If the cost becomes a problem, the known variant is to run +single-partition only for a plan with a floating-point reduction or window, +decided once at build from the expression and recorded in the manifest so that +a heal never re-derives it. *Rejected:* round floats before hashing. A value next to a rounding boundary flips under one unit of noise in the last place, and among millions of values @@ -253,15 +256,13 @@ fix; the float-aggregate heal test and a batch-boundary digest test fail on ## Open questions -1. **Rebuild time under single-partition materialization.** Decides D1 against - D1b. -2. **Nested types.** The spike covers fixed-width, boolean, string and binary +1. **Nested types.** The spike covers fixed-width, boolean, string and binary columns. Lists, structs and maps need a recursive definition. -3. **Types parquet cannot store as given.** A `timestamp[s]` column comes back +2. **Types parquet cannot store as given.** A `timestamp[s]` column comes back as `timestamp[ms]`, so the snapshot's schema differs from the entry's recorded schema. Either the writer refuses such a column, or the entry's schema is recorded from the snapshot. `__row_order` pushes toward the second answer, since the writer adds a column the entry's graph does not have. -4. **Engine upgrades.** Single-partition execution fixes the merge order within +3. **Engine upgrades.** Single-partition execution fixes the merge order within one engine version. Nothing guarantees float results across versions. D4 attributes that case and the remedy stays a rebuild.