perf: stream CSV audits with hash-indexed leakage checks

README.md

@@ -14,6 +14,11 @@ usable in CI.
 See [usage and CI guidance](docs/USAGE_AND_CI.md) for the exact CSV contract,
 JSON output, exit-code handling and known blind spots.
 
+CSV rows are processed incrementally using a training-text hash map and ID/group
+sets, rather than keeping both full datasets in memory. See the
+[algorithm and benchmark notes](docs/ALGORITHM.md) for complexity and a
+reproducible comparison.
+
 The comparison lowercases text, removes punctuation and collapses whitespace.
 It does not detect paraphrases, translations or copied content with substantial
 edits. In a real project, choose groups that match the deployment risk (such as

benchmarks/streaming_audit.py

@@ -0,0 +1,88 @@
+"""Compare streaming and materialised audits on synthetic wide CSV files.
+
+Run from the repository root: python benchmarks/streaming_audit.py
+This measures Python allocations, not process RSS. Dataset creation is excluded.
+"""
+import argparse
+import csv
+import gc
+import json
+import sys
+import tempfile
+import time
+import tracemalloc
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from splitcheck import audit, normalise, read_rows
+
+
+def materialised_audit(train, test):
+    train_rows = read_rows(train, "id", "text")
+    test_rows = read_rows(test, "id", "text")
+    index = {normalise(row["text"]): row["id"] for row in train_rows}
+    overlaps = [
+        {"train_id": index[key], "test_id": row["id"]}
+        for row in test_rows
+        if (key := normalise(row["text"])) in index
+    ]
+    return {
+        "train_rows": len(train_rows), "test_rows": len(test_rows),
+        "exact_text_overlaps": overlaps, "shared_groups": [],
+        "passed": not overlaps,
+    }
+
+
+def measure(function, train, test):
+    gc.collect()
+    tracemalloc.start()
+    started = time.perf_counter()
+    try:
+        result = function(train, test)
+        elapsed = time.perf_counter() - started
+        peak = tracemalloc.get_traced_memory()[1]
+    finally:
+        tracemalloc.stop()
+    return result, {"seconds": elapsed, "peak_python_bytes": peak}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--rows", type=int, default=5000, help="rows per split")
+    parser.add_argument("--columns", type=int, default=8, help="unused payload columns")
+    parser.add_argument("--payload-size", type=int, default=256)
+    args = parser.parse_args()
+    if args.rows < 1 or args.columns < 0 or args.payload_size < 0:
+        parser.error("rows must be positive; columns and payload-size must be nonnegative")
+    with tempfile.TemporaryDirectory() as directory:
+        train, test = Path(directory) / "train.csv", Path(directory) / "test.csv"
+        fields = ["id", "text"] + [f"unused_{i}" for i in range(args.columns)]
+        for path, prefix in [(train, "train"), (test, "test")]:
+            with path.open("w", newline="", encoding="utf-8") as stream:
+                writer = csv.writer(stream)
+                writer.writerow(fields)
+                for i in range(args.rows):
+                    writer.writerow(
+                        [f"{prefix}-{i}", f"{prefix} message {i % 100}"]
+                        + ["x" * args.payload_size] * args.columns
+                    )
+        original, baseline = measure(materialised_audit, train, test)
+        current, streaming = measure(audit, train, test)
+        if original != current:
+            raise AssertionError("streaming report differs from the reference")
+        print(json.dumps({
+            "python": sys.version.split()[0],
+            "rows_per_split": args.rows,
+            "unused_columns": args.columns,
+            "bytes_per_payload_field": args.payload_size,
+            "materialised": baseline,
+            "streaming": streaming,
+            "peak_allocation_reduction_percent": (
+                100 * (1 - streaming["peak_python_bytes"] / baseline["peak_python_bytes"])
+            ),
+            "reports_equal": True,
+        }, indent=2))
+
+
+if __name__ == "__main__":
+    main()

docs/ALGORITHM.md

@@ -0,0 +1,75 @@
+# Streaming leakage audit: algorithm and measurement
+
+## Data structures
+
+The audit reads the training CSV once, normalises each text, and stores its row
+ID in a dictionary keyed by normalised text. Assigning an existing key preserves
+the previous behaviour: the last matching training ID is used in the report.
+An optional set records non-empty training groups.
+
+The test CSV is then read once. Dictionary membership identifies text overlap;
+set membership identifies shared groups. Matches are appended in test-file
+order, while a set deduplicates shared groups before sorting them for the report.
+Unused CSV columns are discarded as iteration advances.
+
+The row iterator validates unique, non-empty IDs with a per-file set. It checks
+the entire input, including rows after the first overlap. A late validation error
+aborts the audit; the CLI does not print a partial JSON report.
+
+`read_rows()` remains available and returns a list for existing callers. The
+`audit()` hot path uses the incremental iterator instead.
+
+## Complexity and limits
+
+For `n` training rows and `m` test rows, dictionary and set operations take
+expected O(n + m) time, plus text normalisation and sorting `g` shared groups in
+O(g log g). Hash-table operations are expected-time, not a worst-case guarantee.
+
+Memory is not constant: the training text index, training group set, per-file
+ID-validation set and output overlaps must still fit in RAM. Its size depends on
+unique text lengths, IDs, groups and number of matches. The change removes the
+additional retention of both complete CSV row lists and their unused columns.
+It is most useful for wide files with large payload fields that are not audited.
+
+## Reproduce the comparison
+
+From the repository root with Python 3.11 or newer:
+
+```bash
+python benchmarks/streaming_audit.py --rows 5000 --columns 8 --payload-size 256
+python -m pytest
+```
+
+The benchmark generates two synthetic CSVs in a temporary directory, then
+compares an equivalent materialised reference against the streaming audit.
+It checks report equality before printing timings and peak Python allocations.
+File generation is excluded from the measurement, and temporary files are cleaned
+up afterwards. No production dataset or network access is used.
+
+`tracemalloc` measures tracked Python allocations, not total process RSS. Timings
+include its overhead and depend on Python, hardware and filesystem caching;
+this is a reproducible microbenchmark, not a production throughput guarantee.
+Use `--columns 0` to compare narrow files, or increase `--rows` to study ID/index
+growth. Keep memory claims tied to the exact reported workload.
+
+### Observed local allocation comparison
+
+A single local run on Python 3.14.5, Windows, on 13 October 2026 used 5,000
+rows per split and 256 characters per unused field:
+
+| Unused columns | Materialised peak bytes | Streaming peak bytes | Reduction |
+| --- | --- | --- | --- |
+| 8 | 28,316,228 | 1,088,533 | 96.2% |
+| 0 | 3,775,481 | 1,075,891 | 71.5% |
+
+Both reports matched the reference exactly. This is a synthetic allocation
+comparison, not a claim about real datasets, production throughput or process RSS.
+The reference uses the same row validation but retains both full row lists.
+
+## Correctness checks
+
+Regression tests cover the streaming path, report equality with a materialised
+reference across seeded random inputs, last-training-ID selection, output order,
+empty groups, custom columns, UTF-8 BOM, missing columns, header-only files,
+late duplicate IDs and incremental iterator validation. Existing example and CLI
+semantics remain unchanged; this does not add semantic-duplicate detection.

docs/USAGE_AND_CI.md

@@ -19,6 +19,10 @@ The file must contain at least one row and a header. It is read using
 `utf-8-sig`, which also accepts a UTF-8 byte-order mark. Change column
 names with `--id-column`, `--text-column` and `--group-column`.
 
+The audit processes rows incrementally rather than retaining both complete CSVs.
+It still stores the training text index, validation IDs and report findings;
+see [Algorithm and measurement](ALGORITHM.md) for the memory boundary and benchmark.
+
 Group values are compared as provided; there is **no normalisation of group
 labels**. For realistic leakage checks, apply a consistent grouping scheme
 upstream (for example the same campaign ID format in both files).

splitcheck.py

@@ -5,42 +5,67 @@
 import csv
 import json
 import re
+from collections.abc import Iterator
+from itertools import chain
 from pathlib import Path
 
 
 def normalise(text: str) -> str:
     return " ".join(re.findall(r"[a-z0-9]+", text.casefold()))
 
 
-def read_rows(path: Path, id_column: str, text_column: str) -> list[dict[str, str]]:
+def _iter_rows(
+    path: Path, id_column: str, text_column: str, group_column=None
+) -> Iterator[dict[str, str]]:
+    """Validate CSV rows incrementally without retaining unused columns."""
     with path.open(newline="", encoding="utf-8-sig") as stream:
-        rows = list(csv.DictReader(stream))
-    if not rows:
-        raise ValueError(f"{path.name} has no rows")
-    missing = {id_column, text_column} - set(rows[0])
-    if missing:
-        raise ValueError(f"{path.name} is missing columns: {sorted(missing)}")
-    ids = [row[id_column].strip() for row in rows]
-    if any(not item for item in ids) or len(ids) != len(set(ids)):
-        raise ValueError(f"{path.name} IDs must be non-empty and unique")
-    return rows
+        reader = csv.DictReader(stream)
+        first = next(reader, None)
+        if first is None:
+            raise ValueError(f"{path.name} has no rows")
+        missing = {id_column, text_column} - set(first)
+        if missing:
+            raise ValueError(f"{path.name} is missing columns: {sorted(missing)}")
+        if group_column and group_column not in first:
+            raise ValueError(f"Both files need group column: {group_column}")
+        seen_ids = set()
+        for row in chain((first,), reader):
+            row_id = row[id_column].strip()
+            if not row_id or row_id in seen_ids:
+                raise ValueError(f"{path.name} IDs must be non-empty and unique")
+            seen_ids.add(row_id)
+            yield row
+
+
+def read_rows(path: Path, id_column: str, text_column: str) -> list[dict[str, str]]:
+    """Return validated rows for callers that explicitly need materialisation."""
+    return list(_iter_rows(path, id_column, text_column))
 
 
 def audit(train: Path, test: Path, id_column="id", text_column="text", group_column=None) -> dict:
-    train_rows = read_rows(train, id_column, text_column)
-    test_rows = read_rows(test, id_column, text_column)
-    train_text = {normalise(row[text_column]): row[id_column] for row in train_rows}
-    overlaps = [
-        {"train_id": train_text[key], "test_id": row[id_column]}
-        for row in test_rows if (key := normalise(row[text_column])) in train_text
-    ]
-    groups = []
-    if group_column:
-        if group_column not in train_rows[0] or group_column not in test_rows[0]:
-            raise ValueError(f"Both files need group column: {group_column}")
-        train_groups = {row[group_column] for row in train_rows if row[group_column]}
-        groups = sorted(train_groups & {row[group_column] for row in test_rows if row[group_column]})
-    return {"train_rows": len(train_rows), "test_rows": len(test_rows),
+    train_text = {}
+    train_groups = set()
+    train_count = 0
+    for row in _iter_rows(train, id_column, text_column, group_column):
+        train_count += 1
+        # Preserve the existing last-training-ID selection for duplicate texts.
+        train_text[normalise(row[text_column])] = row[id_column]
+        if group_column and row[group_column]:
+            train_groups.add(row[group_column])
+
+    overlaps = []
+    shared_groups = set()
+    test_count = 0
+    for row in _iter_rows(test, id_column, text_column, group_column):
+        test_count += 1
+        key = normalise(row[text_column])
+        if key in train_text:
+            overlaps.append({"train_id": train_text[key], "test_id": row[id_column]})
+        if group_column and row[group_column] in train_groups:
+            shared_groups.add(row[group_column])
+
+    groups = sorted(shared_groups)
+    return {"train_rows": train_count, "test_rows": test_count,
             "exact_text_overlaps": overlaps, "shared_groups": groups,
             "passed": not overlaps and not groups}
 

tests/test_iterator_validation.py

@@ -0,0 +1,38 @@
+import csv
+
+import pytest
+
+from splitcheck import _iter_rows, audit
+
+
+def write_rows(path, rows):
+    with path.open("w", newline="", encoding="utf-8") as stream:
+        writer = csv.DictWriter(stream, fieldnames=list(rows[0]))
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+def test_rows_are_validated_as_the_iterator_advances(tmp_path):
+    path = tmp_path / "data.csv"
+    write_rows(path, [{"id": "a", "text": "one"}, {"id": "a", "text": "two"}])
+    rows = _iter_rows(path, "id", "text")
+    assert next(rows) == {"id": "a", "text": "one"}
+    with pytest.raises(ValueError, match="unique"):
+        next(rows)
+
+
+@pytest.mark.parametrize("missing", ["id", "text"])
+def test_missing_required_column(tmp_path, missing):
+    path = tmp_path / "data.csv"
+    row = {"id": "a", "text": "one"}
+    del row[missing]
+    write_rows(path, [row])
+    with pytest.raises(ValueError, match="missing columns"):
+        audit(path, path)
+
+
+def test_whitespace_only_id(tmp_path):
+    path = tmp_path / "data.csv"
+    write_rows(path, [{"id": " ", "text": "one"}])
+    with pytest.raises(ValueError, match="non-empty"):
+        audit(path, path)