Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 8 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 6 additions & 2 deletions docs/repair-plans.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 5 additions & 3 deletions src/freshdata/streaming/_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
78 changes: 78 additions & 0 deletions tests/test_plan_drift_sampling.py
Original file line number Diff line number Diff line change
@@ -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
Loading