diff --git a/plans/ADR-009-digest-stability.md b/plans/ADR-009-digest-stability.md new file mode 100644 index 0000000..5f9763c --- /dev/null +++ b/plans/ADR-009-digest-stability.md @@ -0,0 +1,268 @@ +# 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`, 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 + "`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` + (`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 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 + 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`. + +## 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 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. + +**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 those three ADR-006 decisions +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 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 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 +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 +setting for a different job and accepts that cost knowingly: reproducible +arithmetic is the point, and no setting gives both. + +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 +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 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 + +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 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. + +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 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 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 +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 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 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 + +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 +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. + +## Consequences + +- A float-aggregate entry heals to the same digest, and a pyarrow upgrade no + 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. +- 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. **Nested types.** The spike covers fixed-width, boolean, string and binary + columns. Lists, structs and maps need a recursive definition. +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. +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. 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()