fix: make tallyman_read_csv order-stable via polars, not ibis.row_number

original_row_order is now produced by an order-preserving polars read
(scan_csv -> with_row_index) materialised to a stat-addressed parquet, then read
back via deferred_read_parquet with a top-level order_by. A bare datafusion
ROW_NUMBER() OVER () numbers rows in the nondeterministic arrival order of a
parallel CSV scan (any file over the ~10MB repartition threshold), so it never
pinned file order: the snapshot digest false-drifted run-to-run on the
motivating large-CSV case while the 3-row test stayed green.

The intermediate parquet is keyed on the CSV stat signature (idempotent; skips
the sink if present), written atomically (tmp + rename), with pinned polars
write settings so the bytes are reproducible within an environment.
original_row_order is cast to int64 and placed last to preserve the documented
schema shape; unmapped ibis types raise rather than silently inferring.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: paddymul

src/tallyman_xorq/io.py

@@ -12,7 +12,7 @@
 
 from pathlib import Path
 
-from tallyman_core import data_dir, entry_dir, get_alias, resolve_project
+from tallyman_core import artifacts_dir, data_dir, entry_dir, get_alias, resolve_project
 
 
 class ProjectDataNotFound(FileNotFoundError):
@@ -75,29 +75,124 @@ def read_project_file(rel_path: str, project: str | None = None):
     return deferred_read_parquet(str(path))
 
 
-def tallyman_read_csv(path: str, schema=None, **kwargs):
-    """Read a CSV into a xorq expression with ``original_row_order`` injected.
-
-    Use this instead of ``xo.deferred_read_csv`` for all CSV ingests. It adds
-    a 0-based ``original_row_order`` column (the source file's row sequence)
-    which serves as the canonical total order for snapshot baking — making
-    the entry's result digest byte-stable across builds.
-
-    Because ``original_row_order`` requires a ``RowNumber`` window op, entries
-    built with this function are classified as snapshot-worthy and bake a
-    sorted parquet snapshot on first build. Subsequent reads are fast
-    (parquet scan, not CSV re-parse).
+# Pinned polars parquet-write settings for the ordered-CSV intermediate. Held
+# constant so the bytes — and therefore the snapshot digest — are reproducible
+# run-to-run within an environment. A polars/library upgrade may change the
+# bytes; rebuild-is-fine here per the project's single-user rule.
+_CSV_PARQUET_WRITE = {
+    "compression": "zstd",
+    "compression_level": 3,
+    "row_group_size": 122880,
+    "statistics": True,
+}
+
+# ibis primitive -> polars dtype, for reading a CSV with an explicit schema.
+# Covers the types the MCP documents for tallyman_read_csv; an unmapped type
+# raises rather than silently inferring, so a schema mistake fails loudly.
+_IBIS_TO_POLARS = {
+    "int8": "Int8", "int16": "Int16", "int32": "Int32", "int64": "Int64",
+    "uint8": "UInt8", "uint16": "UInt16", "uint32": "UInt32", "uint64": "UInt64",
+    "float32": "Float32", "float64": "Float64",
+    "string": "String", "bool": "Boolean", "boolean": "Boolean",
+    "date": "Date",
+}
+
+
+def _polars_overrides(schema):
+    """Translate an ibis schema to polars ``schema_overrides`` dtypes."""
+    import polars as pl
+
+    overrides = {}
+    for name, dtype in zip(schema.names, schema.types):
+        key = str(dtype)
+        if key.startswith("timestamp"):
+            overrides[name] = pl.Datetime
+            continue
+        pl_name = _IBIS_TO_POLARS.get(key)
+        if pl_name is None:
+            raise ValueError(
+                f"tallyman_read_csv: column {name!r} has ibis type {key!r}, which has no "
+                f"polars mapping. Supported: {sorted(_IBIS_TO_POLARS)} (+ timestamp*)."
+            )
+        overrides[name] = getattr(pl, pl_name)
+    return overrides
+
+
+def _ordered_csv_parquet(path: str, schema) -> Path:
+    """Materialise *path* to a row-order-stable parquet, keyed on the CSV's stat
+    signature, and return its path (idempotent — skips the write if it exists).
+
+    polars ``scan_csv -> with_row_index -> sink_parquet`` preserves source file
+    order (unlike datafusion's parallel scan, whose row order is nondeterministic
+    above the repartition threshold), so ``original_row_order`` is the true 0..N-1
+    file sequence and the written bytes are reproducible. See
+    ``plans/adr-result-digest-canonical-ordering.md``.
+    """
+    import hashlib
+
+    import polars as pl
+
+    src = Path(path)
+    st = src.stat()
+    schema_sig = repr(sorted(zip(schema.names, [str(t) for t in schema.types]))) if schema is not None else "none"
+    key = hashlib.md5(  # noqa: S324 — naming key, not a security boundary
+        f"{src.resolve()}|{st.st_mtime_ns}|{st.st_size}|{schema_sig}".encode()
+    ).hexdigest()
+
+    out_dir = artifacts_dir(resolve_project(None)) / "csv_ordered"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    target = out_dir / f"{key}.parquet"
+    if target.exists():
+        return target
+
+    lf = pl.scan_csv(
+        str(src),
+        **({"schema_overrides": _polars_overrides(schema), "infer_schema_length": 0} if schema is not None else {}),
+    ).with_row_index("original_row_order")
+    if schema is not None:
+        cols = list(schema.names)
+    else:
+        cols = [c for c in lf.collect_schema().names() if c != "original_row_order"]
+    # original_row_order last (matches the prior mutate-appends-a-column shape)
+    # and cast to int64 (with_row_index yields uint32) for the documented schema.
+    lf = lf.select([*cols, pl.col("original_row_order").cast(pl.Int64)])
+
+    tmp = target.with_suffix(".parquet.tmp")
+    lf.sink_parquet(str(tmp), **_CSV_PARQUET_WRITE)
+    tmp.replace(target)  # atomic — a crash mid-write never leaves a partial at `target`
+    return target
+
+
+def tallyman_read_csv(path: str, schema=None):
+    """Read a CSV into a xorq expression with a stable ``original_row_order``.
+
+    Use this instead of ``xo.deferred_read_csv`` for all CSV ingests. It adds a
+    0-based ``original_row_order`` column holding the source file's true row
+    sequence, which serves as the canonical total order for snapshot baking —
+    making the entry's result digest byte-stable across builds.
+
+    The ordering is produced by an order-preserving polars read
+    (``scan_csv -> with_row_index``) materialised to a stat-addressed parquet,
+    *not* by ``ibis.row_number()``: a bare datafusion ``ROW_NUMBER() OVER ()``
+    numbers rows in the nondeterministic arrival order of a parallel CSV scan
+    (any file over datafusion's ~10 MB repartition threshold), so it would not
+    pin file order at all. See ``plans/adr-result-digest-canonical-ordering.md``.
+
+    The returned expression is a ``deferred_read_parquet`` of that stable file
+    with a top-level ``order_by("original_row_order")``, so it is classified
+    snapshot-worthy and bakes a canonically-ordered parquet snapshot on build;
+    the trailing sort also re-imposes the canonical order even if the parquet
+    scan itself reparallelises.
 
     Args:
         path: Absolute path to the CSV file.
         schema: Optional ibis schema for the columns (same as deferred_read_csv).
-        **kwargs: Forwarded to ``xo.deferred_read_csv``.
+            When omitted, polars infers types.
     """
     import xorq.api as xo
-    import xorq.vendor.ibis as ibis
 
-    t = xo.deferred_read_csv(path, schema=schema, **kwargs)
-    return t.mutate(original_row_order=ibis.row_number()).order_by("original_row_order")
+    parquet_path = _ordered_csv_parquet(path, schema)
+    return xo.deferred_read_parquet(str(parquet_path)).order_by("original_row_order")
 
 
 def _reconstructing_source_digest(proj: str, rel_path: str) -> str | None: