diff --git a/docs/streaming.md b/docs/streaming.md index 8db5cc2..49c1b31 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -154,7 +154,8 @@ Each batch is compared against the locked schema baseline and the running state, - a column's dtype changes; - its missing ratio jumps sharply (`drift_missing_jump`); - its cardinality explodes (`drift_cardinality_factor`); -- its numeric mean shifts past `drift_zscore` σ from the running mean. +- its numeric mean shifts past `drift_zscore` σ from the running mean, or moves at all + for a column that has been constant so far. ## Input formats @@ -231,7 +232,8 @@ fewer rows are processed than requested. - Streaming mode is **micro-batch**, not true row-by-row real time. - **Global** cross-batch duplicate detection is limited: by default duplicates are scoped within a batch; enabling `global_duplicates` uses a *bounded recent-window* - that can miss duplicates older than the window. + (the `window_size` most recently seen distinct rows) that can miss duplicates + older than the window. - Medians/quantiles are **approximate** (reservoir-sampled), and the top-k category summary is approximate when a column saturates `max_categories`. - Kafka and Arrow Flight are **optional** integrations. diff --git a/src/freshdata/streaming/_cleaner.py b/src/freshdata/streaming/_cleaner.py index 659bf48..400ae6c 100644 --- a/src/freshdata/streaming/_cleaner.py +++ b/src/freshdata/streaming/_cleaner.py @@ -12,15 +12,17 @@ from __future__ import annotations import dataclasses +from collections import OrderedDict from collections.abc import Iterable, Iterator from typing import TYPE_CHECKING, Any +import numpy as np import pandas as pd if TYPE_CHECKING: from ..context import ContextPolicy -from .._util import exceeds_float64_exact, fill_na_exact +from .._util import FLOAT64_EXACT_INT, exceeds_float64_exact, fill_na_exact from ..cleaner import run_pipeline from ..config import CleanConfig, merge_options from ..engine.context import infer_role @@ -75,7 +77,9 @@ def __init__(self, *, config: CleanConfig | None = None, ) self._roles: dict[str, str] = {} self.report_: CleanReport | None = None - self._seen_hashes: set[int] = set() # recent-window cross-batch dedup + # Recent-window cross-batch dedup: row hashes in least- to most-recently-seen + # order, capped at ``window_size`` (see ``_dedup_window``). + self._seen_hashes: OrderedDict[int, None] = OrderedDict() self._n_imputed = 0 self._n_deferred = 0 self._gate_failures = 0 @@ -373,12 +377,19 @@ def _lock_roles(self, df: pd.DataFrame) -> None: self._roles[name] = infer_role(name, df[col], self.config) def _dedup_window(self, df: pd.DataFrame, report: CleanReport) -> pd.DataFrame: - # ponytail: bounded recent-window dedup (cap = window_size), not true global. - hashes = pd.util.hash_pandas_object(df, index=False).to_numpy() - keep = [h not in self._seen_hashes for h in hashes] - for h, k in zip(hashes, keep): - if k and len(self._seen_hashes) < self.scfg.window_size: - self._seen_hashes.add(int(h)) + # Bounded recent-window dedup, not true global: a row is dropped when it + # repeats one of the ``window_size`` most recently seen distinct rows of + # earlier batches. Within-batch duplicates are left to the pipeline. + hashes = _row_hashes(df) + seen = self._seen_hashes + keep = [h not in seen for h in hashes] + for h in hashes: + if h in seen: + seen.move_to_end(h) # a repeat counts as recent again + else: + seen[h] = None + while len(seen) > self.scfg.window_size: + seen.popitem(last=False) # evict the least recently seen row removed = len(df) - sum(keep) if removed: report.add("duplicates", f"removed {removed} cross-batch duplicate row(s) " @@ -507,3 +518,33 @@ def _preserve(self, df: pd.DataFrame, col: str, miss: int, report: CleanReport, def _round(value: float | None) -> float | None: return round(value, 2) if value is not None else None + + +def _row_hashes(df: pd.DataFrame) -> list[int]: + """Per-row hashes that are stable when a numeric column's dtype flips. + + ``hash_pandas_object`` hashes the dtype-specific bytes, so ``1`` (int64) and + ``1.0`` (float64, e.g. an int column promoted by a missing value) differ. + Numeric non-bool columns, nullable ones included, are hashed as float64 + (missing -> NaN, ``-0.0`` -> ``0.0``). An integer column holding a value + beyond ±2**53 keeps its own dtype, because float64 would merge distinct values. + """ + columns: dict[int, pd.Series] = {} + changed = False + for i in range(df.shape[1]): + s = df.iloc[:, i] + kind = getattr(s.dtype, "kind", None) + if kind in ("i", "u", "f") and (kind == "f" or _ints_fit_float64(s)): + values = s.to_numpy(dtype="float64", na_value=np.nan) + 0.0 + s = pd.Series(values, index=df.index) + changed = True + columns[i] = s + frame = pd.DataFrame(columns, index=df.index) if changed else df + return pd.util.hash_pandas_object(frame, index=False).tolist() + + +def _ints_fit_float64(s: pd.Series) -> bool: + present = s.dropna() + if present.empty: + return True + return int(present.max()) <= FLOAT64_EXACT_INT and int(present.min()) >= -FLOAT64_EXACT_INT diff --git a/src/freshdata/streaming/_config.py b/src/freshdata/streaming/_config.py index e68e6d4..6a7516d 100644 --- a/src/freshdata/streaming/_config.py +++ b/src/freshdata/streaming/_config.py @@ -40,6 +40,9 @@ class StreamingCleanConfig: global_duplicates: When False (default), duplicate removal is scoped **within each batch** only — cross-batch duplicate detection would require unbounded state. + When True, a row is also dropped if it repeats one of the ``window_size`` + most recently seen distinct rows from earlier batches (least recently + seen rows are evicted first). drift_missing_jump: Absolute jump in a column's missing ratio (batch vs. running) that flags missing-rate drift. @@ -48,7 +51,8 @@ class StreamingCleanConfig: count flags a cardinality explosion. drift_zscore: |batch mean − running mean| / running std above this flags a numeric - distribution shift. + distribution shift. For a column that has been constant so far (running + std 0), any change of the batch mean flags it. seed: Seed for reservoir sampling, for reproducible approximate quantiles. """ diff --git a/src/freshdata/streaming/_drift.py b/src/freshdata/streaming/_drift.py index 124063a..fa81783 100644 --- a/src/freshdata/streaming/_drift.py +++ b/src/freshdata/streaming/_drift.py @@ -112,12 +112,24 @@ def _distribution_drift(s, cstate, config, findings) -> None: if not (is_numeric_dtype(s) and not is_bool_dtype(s)): return snap = cstate.numeric_snapshot() - if snap.count < 2 or snap.std <= 0: + if snap.count < 2: return nonnull = pd.to_numeric(s, errors="coerce").dropna() if nonnull.empty: return - z = abs(float(nonnull.mean()) - snap.mean) / snap.std + batch_mean = float(nonnull.mean()) + if snap.std <= 0: + # Constant so far: the z-score is undefined (any move is infinite σ), so + # flag a mean that leaves the constant by more than a float tolerance. + if abs(batch_mean - snap.mean) > 1e-9 * max(1.0, abs(snap.mean)): + findings.append(DriftFinding( + "distribution", cstate.name, + f"mean of '{cstate.name}' moved to {batch_mean:.4g} from a column " + f"that was constant at {snap.mean:.4g}", + risk="high", + )) + return + z = abs(batch_mean - snap.mean) / snap.std if z > config.drift_zscore: findings.append(DriftFinding( "distribution", cstate.name, diff --git a/tests/test_streaming_dedup_drift.py b/tests/test_streaming_dedup_drift.py new file mode 100644 index 0000000..2113cca --- /dev/null +++ b/tests/test_streaming_dedup_drift.py @@ -0,0 +1,148 @@ +"""Regression tests for streaming cross-batch dedup (#292, #293) and drift (#294).""" + +from __future__ import annotations + +import pandas as pd + +import freshdata as fd + + +def _keys_batch(keys): + return pd.DataFrame({"k": keys, "s": [f"x{i}" for i in keys]}) + + +def _run(cleaner, batches): + return [cleaner.clean_batch(b)[0] for b in batches] + + +# -- #292: the window keeps the most recent rows, not the first ones ------------------- + + +def test_duplicates_of_recent_rows_are_removed_after_window_fills(): + cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False) + outs = _run(cleaner, [_keys_batch([1, 2]), _keys_batch([3, 4]), _keys_batch([3, 4])]) + assert outs[2]["k"].tolist() == [] + + +def test_duplicates_after_more_than_window_size_distinct_rows(): + cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=3, verbose=False) + batches = [_keys_batch([1, 2, 3]), _keys_batch([4, 5, 6]), _keys_batch([7, 8]), + _keys_batch([6, 7, 8])] + outs = _run(cleaner, batches) + assert outs[3]["k"].tolist() == [] + + +def test_oldest_rows_are_evicted_and_window_stays_bounded(): + cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False) + outs = _run(cleaner, [_keys_batch([1, 2]), _keys_batch([3, 4]), _keys_batch([1])]) + assert outs[2]["k"].tolist() == [1] # 1 fell out of the 2-row window + assert len(cleaner._seen_hashes) <= 2 + + +def test_repeat_refreshes_recency(): + cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False) + outs = _run(cleaner, [ + _keys_batch([1, 2]), + _keys_batch([1]), # duplicate: removed, and 1 becomes most recent + _keys_batch([3]), # evicts 2, the least recently seen + _keys_batch([1]), + _keys_batch([2]), + ]) + assert [len(o) for o in outs] == [2, 0, 1, 0, 1] + + +def test_batch_larger_than_window_keeps_only_latest_rows(): + cleaner = fd.StreamingCleaner(global_duplicates=True, window_size=2, verbose=False) + outs = _run(cleaner, [_keys_batch([1, 2, 3, 4]), _keys_batch([1, 3, 4])]) + assert outs[1]["k"].tolist() == [1] + assert len(cleaner._seen_hashes) == 2 + + +# -- #293: hashing survives int64 <-> float64 flips ------------------------------------ + + +def test_int_row_repeats_in_nan_promoted_float_batch(): + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"k": [1, 2], "v": [10, 20]})) + out, report = cleaner.clean_batch(pd.DataFrame({"k": [1, 3], "v": [10, None]})) + assert out["k"].tolist() == [3] + assert report.duplicates_removed == 1 + + +def test_float_row_repeats_in_later_int_batch(): + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"k": [1, 2], "v": [10.0, None]})) + out, _ = cleaner.clean_batch(pd.DataFrame({"k": [2, 5], "v": [20, 30]})) + assert out["k"].tolist() == [2, 5] # (2, NaN) != (2, 20) + out, _ = cleaner.clean_batch(pd.DataFrame({"k": [1, 6], "v": [10, 60]})) + assert out["k"].tolist() == [6] + + +def test_nullable_integer_matches_plain_float(): + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"v": pd.array([1, None, 3], dtype="Int64")})) + out, _ = cleaner.clean_batch(pd.DataFrame({"v": [3.0, 4.0]})) + assert out["v"].tolist() == [4.0] + + +def test_negative_zero_matches_integer_zero(): + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"k": ["a", "b"], "v": [0, 1]})) + out, _ = cleaner.clean_batch(pd.DataFrame({"k": ["a", "c"], "v": [-0.0, 2.5]})) + assert out["k"].tolist() == ["c"] + + +def test_distinct_numeric_values_are_not_merged(): + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"k": ["a"], "v": [1]})) + out, _ = cleaner.clean_batch(pd.DataFrame({"k": ["a"], "v": [1.5]})) + assert len(out) == 1 + + +def test_integers_beyond_float64_precision_stay_distinct(): + big = 2**53 + cleaner = fd.StreamingCleaner(global_duplicates=True, verbose=False) + cleaner.clean_batch(pd.DataFrame({"id": [big + 1]})) + out, _ = cleaner.clean_batch(pd.DataFrame({"id": [big]})) + assert out["id"].tolist() == [big] + + +# -- #294: drift on a column that was constant ------------------------------------------ + + +def _drift_actions(report): + return [a for a in report.actions if a.step == "drift"] + + +def test_constant_column_jump_reports_distribution_drift(): + cleaner = fd.StreamingCleaner(verbose=False) + for val in (5, 5, 5, 1_000_000): + _, report = cleaner.clean_batch(pd.DataFrame({"x": [val] * 50, "y": list(range(50))})) + assert report.streaming["schema_drift_detected"] is True + actions = _drift_actions(report) + assert [a.column for a in actions] == ["x"] + assert actions[0].risk == "high" + assert cleaner.state.drift_log[-1]["kind"] == "distribution" + + +def test_constant_column_unchanged_reports_no_drift(): + cleaner = fd.StreamingCleaner(verbose=False) + for _ in range(4): + _, report = cleaner.clean_batch(pd.DataFrame({"x": [5.0] * 50, "y": list(range(50))})) + assert _drift_actions(report) == [] + assert cleaner.state.drift_log == [] + + +def test_constant_column_within_float_tolerance_reports_no_drift(): + cleaner = fd.StreamingCleaner(verbose=False) + cleaner.clean_batch(pd.DataFrame({"x": [0.1] * 10, "y": list(range(10))})) + _, report = cleaner.clean_batch( + pd.DataFrame({"x": [0.1 + 1e-15] * 10, "y": list(range(10))})) + assert _drift_actions(report) == [] + + +def test_single_prior_value_does_not_report_drift(): + cleaner = fd.StreamingCleaner(verbose=False) + cleaner.clean_batch(pd.DataFrame({"x": [5], "y": ["a"]})) + _, report = cleaner.clean_batch(pd.DataFrame({"x": [9], "y": ["a"]})) + assert _drift_actions(report) == []