diff --git a/CHANGELOG.md b/CHANGELOG.md index 6868390..d62ad8b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Documentation + +- `StreamingCleanConfig.window_size` was documented as sizing "rolling + statistics and the rolling trust score". It does neither: its only effect is + to bound the cross-batch duplicate window, and the rolling trust score is + sized by `rolling_trust_window`. Docstring corrected; no behaviour change. +- `docs/repair-plans.md` now states that the `FrameSignature` content sample is + the first 512 rows, so a change beyond row 512 that preserves row count, + column names and dtypes is not detected by drift refusal. +- The README's "Native Polars DataFrames" section now states what + `docs/fallback-matrix.md` already did: with default options every native + engine delegates the whole pipeline to pandas, and the fully native path is + `strategy="conservative"` with `fix_dtypes=False`. + + ### Fixed - A domain regex rule no longer fails a valid code because its column is `float64`. A numeric code column with one blank cell loads from CSV as diff --git a/README.md b/README.md index a867875..7c63c0a 100644 --- a/README.md +++ b/README.md @@ -224,6 +224,14 @@ cleaned = fd.clean(df) ### 2. Native Polars DataFrames Pass a Polars DataFrame, get a Polars DataFrame back with zero pandas boilerplate: + +> **With default options the work still runs on pandas.** The default +> `strategy="balanced"` runs the accuracy-first decision engine, which is +> evaluated by the pandas backend, so every native engine delegates the whole +> pipeline to pandas and records it on `report.fallback_events`. You get a +> Polars frame back, but not native Polars execution. The fully native path is +> `strategy="conservative"` with `fix_dtypes=False`. See +> [docs/fallback-matrix.md](docs/fallback-matrix.md). ```python import polars as pl import freshdata as fd diff --git a/docs/repair-plans.md b/docs/repair-plans.md index ea81e1c..675277e 100644 --- a/docs/repair-plans.md +++ b/docs/repair-plans.md @@ -86,8 +86,12 @@ everything non-trivial stays `pending`. ## Drift refusal A plan remembers the frame it was built for (`FrameSignature`: row count, -column names+dtypes, content sample). Applying it to different data refuses -by default: +column names+dtypes, content sample). The content sample is the **first 512 +rows** (`_SIGNATURE_SAMPLE_ROWS`), not the whole frame: a change beyond row +512 that leaves the row count, column names and dtypes intact is **not** +detected. The fingerprint is deliberately cheap; use it as a guard against +applying a plan to the wrong data, not as a proof the data is unchanged. +Applying a plan to different data refuses by default: ```python fd.apply_plan(other_df, rp) # raises fd.PlanDriftError diff --git a/src/freshdata/streaming/_config.py b/src/freshdata/streaming/_config.py index 6a7516d..cac6f9a 100644 --- a/src/freshdata/streaming/_config.py +++ b/src/freshdata/streaming/_config.py @@ -19,9 +19,11 @@ class StreamingCleanConfig: Parameters ---------- window_size: - Size of the recent-window used for rolling statistics and the rolling - trust score. The *caller* controls how big each batch is; this only - bounds how much recent history influences "recent-window" reporting. + Bound on the cross-batch duplicate window: the number of most recently + seen distinct rows retained when ``global_duplicates`` is enabled. A + duplicate older than the window is not detected. This is its only + effect -- it does **not** size rolling statistics, and the rolling + trust score is sized by ``rolling_trust_window`` instead. warmup_batches: Number of leading batches during which the cleaner only repairs representation and *collects* statistics — it defers statistical diff --git a/tests/test_plan_drift_sampling.py b/tests/test_plan_drift_sampling.py new file mode 100644 index 0000000..6df78d4 --- /dev/null +++ b/tests/test_plan_drift_sampling.py @@ -0,0 +1,78 @@ +"""The frame fingerprint behind drift refusal samples the HEAD, not the frame. + +``docs/repair-plans.md`` describes ``FrameSignature`` as "row count, column +names+dtypes, content sample". That wording is accurate but incomplete in a way +that matters: the sample is the first ``_SIGNATURE_SAMPLE_ROWS`` (512) rows, so +a change *after* row 512 that preserves the row count, the column names and the +dtypes does not trip ``PlanDriftError``. + +That is a deliberate design choice -- the fingerprint is documented as "cheap" +and is a guard against applying a plan to the *wrong data*, not a proof the +data is unchanged. These tests pin the boundary so the documented claim is +backed by execution rather than prose, and so a future change to the sampling +strategy has to come past a failing test. +""" + +from __future__ import annotations + +import pandas as pd +import pytest + +import freshdata as fd +from freshdata.repairplan import _SIGNATURE_SAMPLE_ROWS, compute_frame_signature + + +def _frame(n: int) -> pd.DataFrame: + # A tie-free majority so the plan carries a stable semantic action. + lower = max(1, n // 4) + return pd.DataFrame({"country": ["USA"] * (n - lower) + ["usa"] * lower}) + + +def _plan(df: pd.DataFrame): + return fd.suggest_plan(df, semantic_mode="review") + + +def test_the_documented_sample_size_is_the_one_the_code_uses(): + assert _SIGNATURE_SAMPLE_ROWS == 512 + + +def test_a_change_inside_the_head_sample_is_refused(): + df = _frame(1000) + plan = _plan(df) + drifted = df.copy() + drifted.loc[10, "country"] = "COMPLETELY-DIFFERENT" + with pytest.raises(fd.PlanDriftError): + fd.apply_plan(drifted, plan) + + +def test_a_change_beyond_the_head_sample_is_not_detected(): + """Documented limitation, pinned deliberately -- not an endorsement.""" + df = _frame(1000) + plan = _plan(df) + drifted = df.copy() + drifted.loc[900, "country"] = "COMPLETELY-DIFFERENT" + fd.apply_plan(drifted, plan) # no PlanDriftError + + +def test_the_boundary_sits_exactly_at_the_sample_size(): + df = _frame(_SIGNATURE_SAMPLE_ROWS + 10) + base = compute_frame_signature(df) + + last_seen = df.copy() + last_seen.loc[_SIGNATURE_SAMPLE_ROWS - 1, "country"] = "CHANGED" + assert compute_frame_signature(last_seen).sample_hash != base.sample_hash + + first_unseen = df.copy() + first_unseen.loc[_SIGNATURE_SAMPLE_ROWS, "country"] = "CHANGED" + assert compute_frame_signature(first_unseen).sample_hash == base.sample_hash + + +def test_row_count_and_column_changes_are_still_caught_beyond_the_sample(): + """The sample is only one of three components; the other two still apply.""" + df = _frame(_SIGNATURE_SAMPLE_ROWS + 10) + base = compute_frame_signature(df) + + assert compute_frame_signature(df.iloc[:-1]).n_rows != base.n_rows + + renamed = df.rename(columns={"country": "nation"}) + assert compute_frame_signature(renamed).columns_hash != base.columns_hash