From 300ab5317bc0494359795b3e66332b471de6940e Mon Sep 17 00:00:00 2001 From: Kevin Costner <120246174+kevincostner17@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:57:35 +0530 Subject: [PATCH] fix(cdc): null-key ordering checks, zero stale_after, epoch event times, replay_risk docs - Null CDC keys (#325): the per-key running watermark used groupby(key).cummax(), which drops null keys, so rows with a null key were never classified as out_of_order or late. Group with dropna=False so null-key rows form one group of their own (works on pandas 1.5 and 2.x), and report them as a new `missing_key` warning (row labels as samples). The trust penalty formula is unchanged. - Zero stale_after (#326): the freshness penalty divided the age by stale_after, raising ZeroDivisionError for "0s". With stale_after == 0 the penalty is now the limit of age / stale_after: 1.0 for any positive age and 0.0 otherwise, matching the `stale` defect (age > stale_after). A negative stale_after now raises ValueError. - Integer epoch event times (#327): numeric event-time columns went through pd.to_datetime without a unit and were read as nanoseconds, so Debezium ts_ms values landed in 1970. Parse them with the shared parse_timestamps helper, which infers the unit from magnitude as clean_timeseries does, and add `event_time_unit` ("s", "ms", "us", "ns") to set it explicitly. String and datetime columns are unchanged. - replay_risk (#328): behaviour kept. The replay_threshold docstring now says at least one duplicate_key row is required, so a late-only batch never raises replay_risk. Closes #325 Closes #326 Closes #327 Closes #328 --- src/freshdata/cdc.py | 89 ++++++++++++--- tests/test_cdc_profile_fixes.py | 188 ++++++++++++++++++++++++++++++++ 2 files changed, 259 insertions(+), 18 deletions(-) create mode 100644 tests/test_cdc_profile_fixes.py diff --git a/src/freshdata/cdc.py b/src/freshdata/cdc.py index a2431122..583ed7e7 100644 --- a/src/freshdata/cdc.py +++ b/src/freshdata/cdc.py @@ -2,11 +2,11 @@ Change-data-capture and event streams fail in ways that are *not* missing values: records arrive **stale**, **late** (past the watermark), **out of order**, with -**duplicate CDC keys**, **invalid operation codes**, a **missing event time**, or -as a **replay-risk** batch. Treating those as nulls hides them. ``cdc_profile`` is -a read-only profiler that classifies these freshness/ordering defects separately -from completeness defects and reports enterprise trust penalties for freshness, -ordering, and CDC integrity. +**duplicate CDC keys**, a **missing CDC key**, **invalid operation codes**, a +**missing event time**, or as a **replay-risk** batch. Treating those as nulls +hides them. ``cdc_profile`` is a read-only profiler that classifies these +freshness/ordering defects separately from completeness defects and reports +enterprise trust penalties for freshness, ordering, and CDC integrity. >>> import freshdata as fd >>> rep = fd.cdc_profile(df, event_time="event_ts", key="entity_id", @@ -27,7 +27,7 @@ import pandas as pd -from .streaming._timeseries import coerce_datetimes, to_timedelta +from .streaming._timeseries import TIMESTAMP_UNITS, parse_timestamps, to_timedelta __all__ = ["CDCDefect", "CDCReport", "cdc_profile"] @@ -39,6 +39,7 @@ "late", "out_of_order", "duplicate_key", + "missing_key", "invalid_operation", "missing_event_time", "replay_risk", @@ -156,14 +157,18 @@ def _sample_keys( return tuple(str(v) for v in vals) -def _parse_event_time(raw: pd.Series) -> pd.Series: +def _parse_event_time(raw: pd.Series, unit: str | None = None) -> pd.Series: """Parse the event-time column; unparseable values become ``NaT``. Datetime columns pass through. Mixed UTC offsets (e.g. across a DST change) or a mix of naive and offset-aware values are normalised to UTC, reading naive values as UTC (see :func:`~freshdata.streaming._timeseries.coerce_datetimes`). + Numeric columns are epoch values in *unit*, inferred from their magnitude + when *unit* is ``None``, as ``fd.clean_timeseries`` does + (see :func:`~freshdata.streaming._timeseries.parse_timestamps`). """ - return coerce_datetimes(raw) + parsed, _ = parse_timestamps(raw, unit) + return parsed def _to_event_tz(value: object, tz: Any) -> pd.Timestamp: @@ -182,9 +187,13 @@ def _to_event_tz(value: object, tz: Any) -> pd.Timestamp: def _running_max(ts: pd.Series, key_series: pd.Series | None) -> pd.Series: - """Per-key (or global) running max of event time, in arrival order.""" + """Per-key (or global) running max of event time, in arrival order. + + Rows with a null key form one group of their own (``dropna=False``), so + they are still checked for ordering and lateness. + """ if key_series is not None: - return ts.groupby(key_series, sort=False).cummax() + return ts.groupby(key_series, sort=False, dropna=False).cummax() return ts.cummax() @@ -229,6 +238,23 @@ def _ordering_defects( return {"out_of_order": n_ooo, "late": n_late} +def _missing_key_defect(df: pd.DataFrame, key: str, defects: list[CDCDefect]) -> None: + """Flag rows whose CDC key is null (a warning; the trust penalty is unchanged).""" + null_key_mask = df[key].isna() + n_null_key = int(null_key_mask.sum()) + if n_null_key: + defects.append( + CDCDefect( + kind="missing_key", + level="warning", + n_rows=n_null_key, + rationale=f"{key!r} is null; these rows are checked for ordering as one group", + # The key is null, so sample the row labels instead. + sample_keys=_sample_keys(null_key_mask, df, None), + ) + ) + + def _freshness( ts: pd.Series, now: pd.Timestamp | None, @@ -271,6 +297,7 @@ def cdc_profile( now: object | None = None, stale_after: object | None = None, replay_threshold: float = 0.2, + event_time_unit: str | None = None, ) -> CDCReport: """Classify CDC / event-time defects without mutating *df*. @@ -280,10 +307,13 @@ def cdc_profile( The change/event frame, in arrival order. event_time: Name of the event-timestamp column. Parsed with ``errors="coerce"``; - unparseable/empty values become ``missing_event_time`` defects. + unparseable/empty values become ``missing_event_time`` defects. A + numeric column holds epoch values (see ``event_time_unit``). key: Optional CDC key column (entity id). Enables per-key ordering and - duplicate-key detection. + duplicate-key detection. Rows with a null key are reported as a + ``missing_key`` warning and checked for ordering as one group of their + own. watermark: Optional explicit low watermark (timestamp). Records strictly before it are reported as ``late``. When omitted a *running* per-key watermark is @@ -302,9 +332,18 @@ def cdc_profile( defaults to the current UTC time; pass it explicitly for determinism. Naive ``now`` / ``watermark`` values and naive event times are read as UTC; references are converted to the event times' zone before comparing. + ``stale_after`` must not be negative; ``0`` means any positive age is + stale (freshness penalty 1.0). replay_threshold: - Fraction of (duplicate + late) rows above which a ``replay_risk`` batch - warning is raised. + Fraction of (duplicate + late) rows at or above which a ``replay_risk`` + batch warning is raised. The warning needs at least one + ``duplicate_key`` row: a batch that is only late (or only out of + order) never raises ``replay_risk``, however high its late share. + event_time_unit: + Epoch unit (``"s"``, ``"ms"``, ``"us"`` or ``"ns"``) of a *numeric* + ``event_time`` column, such as Debezium's ``ts_ms``. ``None`` (default) + infers it from the magnitude of the values, as ``fd.clean_timeseries`` + does. Ignored for string and datetime columns. Returns ------- @@ -316,12 +355,18 @@ def cdc_profile( for col in (event_time, key, sequence, operation_col): if col is not None and col not in df.columns: raise KeyError(f"column {col!r} not found in frame") + if event_time_unit is not None and event_time_unit not in TIMESTAMP_UNITS: + raise ValueError( + f"event_time_unit must be one of {TIMESTAMP_UNITS}, got {event_time_unit!r}" + ) n_rows = len(df) lateness_td = to_timedelta(lateness) or pd.Timedelta(0) stale_after_td = to_timedelta(stale_after) + if stale_after_td is not None and stale_after_td < pd.Timedelta(0): + raise ValueError(f"stale_after must not be negative, got {stale_after!r}") - ts = _parse_event_time(df[event_time]) + ts = _parse_event_time(df[event_time], event_time_unit) event_tz = ts.dt.tz now_ts = _to_event_tz(now, event_tz) if now is not None else None @@ -361,9 +406,10 @@ def cdc_profile( else: counts = _ordering_defects(df, ts, key, lateness_td, defects) - # 3) duplicate CDC keys (replayed/duplicated changes). + # 3) missing and duplicate CDC keys (replayed/duplicated changes). n_dup = 0 if key is not None: + _missing_key_defect(df, key, defects) subset = [key, event_time] + ([sequence] if sequence else []) dup_mask = df.duplicated(subset=subset, keep=False) n_dup = int(dup_mask.sum()) @@ -400,7 +446,8 @@ def cdc_profile( # 5) freshness / stale batch. freshness_seconds = _freshness(ts, now_ts, stale_after_td, defects) - # 6) replay-risk batch heuristic. + # 6) replay-risk batch heuristic. A replay re-delivers changes, so at least + # one duplicate-key row is required; late rows alone never trigger it. n_late = counts.get("late", 0) if n_rows and n_dup and (n_dup + n_late) / n_rows >= replay_threshold: defects.append( @@ -450,7 +497,13 @@ def _trust_penalties( ordering = min(1.0, (n_ooo + n_late) / denom) cdc = min(1.0, (n_badop + n_dup + n_missing) / denom) if freshness_seconds is not None and stale_after_td is not None: - freshness = min(1.0, max(0.0, freshness_seconds / stale_after_td.total_seconds())) + stale_after_seconds = stale_after_td.total_seconds() + if stale_after_seconds > 0: + freshness = min(1.0, max(0.0, freshness_seconds / stale_after_seconds)) + else: + # The limit of age / stale_after as stale_after -> 0: any positive age + # is fully stale, and a zero or negative age is not stale at all. + freshness = 1.0 if freshness_seconds > 0 else 0.0 else: freshness = 0.0 return {"freshness": round(freshness, 4), "ordering": round(ordering, 4), "cdc": round(cdc, 4)} diff --git a/tests/test_cdc_profile_fixes.py b/tests/test_cdc_profile_fixes.py new file mode 100644 index 00000000..3cb9a459 --- /dev/null +++ b/tests/test_cdc_profile_fixes.py @@ -0,0 +1,188 @@ +"""Regression tests for ``fd.cdc_profile`` fixes (#325, #326, #327, #328).""" + +from __future__ import annotations + +import pandas as pd +import pytest + +import freshdata as fd + + +def _counts(rep: fd.CDCReport) -> dict[str, int]: + return {d.kind: d.n_rows for d in rep.defects} + + +# -- #325: rows with a null key are still ordering-checked ---------------------- + + +def test_null_key_rows_are_checked_for_lateness(): + ts = pd.to_datetime( + ["2024-01-01 00:10", "2024-01-01 00:00", "2024-01-01 00:20", "2024-01-01 00:00"] + ) + df = pd.DataFrame({"ts": ts, "k": [None, None, "b", "b"]}) + rep = fd.cdc_profile(df, event_time="ts", key="k", now="2024-01-02") + counts = _counts(rep) + assert counts["late"] == 2 # row 1 (null key) and row 3 (key b) + assert counts["missing_key"] == 2 + missing = next(d for d in rep.defects if d.kind == "missing_key") + assert missing.level == "warning" + assert missing.sample_keys == ("0", "1") # row labels, since the key is null + # The trust penalty formula is unchanged: missing keys do not add to it. + assert rep.trust_penalties["cdc"] == 0.0 + assert rep.trust_penalties["ordering"] == 0.5 + + +def test_null_keys_form_their_own_group_when_interleaved(): + ts = pd.to_datetime( + ["2024-01-01 00:10", "2024-01-01 09:00", "2024-01-01 00:09", "2024-01-01 09:01"] + ) + # NaN and None are both null and share one group; key "a" is in order. + df = pd.DataFrame({"ts": ts, "k": [None, "a", float("nan"), "a"]}) + rep = fd.cdc_profile(df, event_time="ts", key="k", lateness="5m", now="2024-01-02") + counts = _counts(rep) + assert counts.get("out_of_order") == 1 + assert "late" not in counts + assert counts["missing_key"] == 2 + + +def test_missing_key_alone_does_not_fail_the_gate(): + df = pd.DataFrame( + {"ts": pd.to_datetime(["2024-01-01 00:00", "2024-01-01 00:01"]), "k": ["a", None]} + ) + rep = fd.cdc_profile(df, event_time="ts", key="k", now="2024-01-01 00:02") + assert _counts(rep) == {"missing_key": 1} + assert rep.passed + + +def test_missing_key_reported_with_explicit_watermark(): + df = pd.DataFrame( + {"ts": pd.to_datetime(["2024-01-01 00:00", "2024-01-01 00:01"]), "k": [None, "a"]} + ) + rep = fd.cdc_profile( + df, event_time="ts", key="k", watermark="2024-01-01 00:01", now="2024-01-01 00:02" + ) + assert _counts(rep) == {"late": 1, "missing_key": 1} + + +def test_no_missing_key_defect_without_nulls_or_key(): + df = pd.DataFrame({"ts": pd.to_datetime(["2024-01-01 00:00"]), "k": ["a"]}) + assert "missing_key" not in _counts( + fd.cdc_profile(df, event_time="ts", key="k", now="2024-01-02") + ) + assert "missing_key" not in _counts(fd.cdc_profile(df, event_time="ts", now="2024-01-02")) + + +# -- #326: zero stale_after ----------------------------------------------------- + + +@pytest.mark.parametrize("stale_after", ["0s", 0, pd.Timedelta(0)]) +def test_zero_stale_after_marks_any_positive_age_stale(stale_after): + df = pd.DataFrame({"ts": pd.to_datetime(["2024-01-01"])}) + rep = fd.cdc_profile(df, event_time="ts", now="2024-01-02", stale_after=stale_after) + assert "stale" in _counts(rep) + assert rep.trust_penalties["freshness"] == 1.0 + + +def test_zero_stale_after_with_zero_age_is_not_stale(): + df = pd.DataFrame({"ts": pd.to_datetime(["2024-01-02"])}) + rep = fd.cdc_profile(df, event_time="ts", now="2024-01-02", stale_after="0s") + assert rep.freshness_seconds == 0.0 + assert "stale" not in _counts(rep) + assert rep.trust_penalties["freshness"] == 0.0 + + +def test_negative_stale_after_raises(): + df = pd.DataFrame({"ts": pd.to_datetime(["2024-01-01"])}) + with pytest.raises(ValueError, match="stale_after"): + fd.cdc_profile(df, event_time="ts", now="2024-01-02", stale_after="-1s") + + +# -- #327: integer epoch event times -------------------------------------------- + + +def test_epoch_milliseconds_are_inferred(): + df = pd.DataFrame({"ts_ms": [1704067200000, 1704067260000], "k": ["a", "b"]}) + rep = fd.cdc_profile(df, event_time="ts_ms", key="k", now="2024-01-01 00:05", stale_after="1h") + assert rep.freshness_seconds == pytest.approx(240.0) + assert rep.defects == [] + + +@pytest.mark.parametrize( + ("values", "unit"), + [ + ([1704067200, 1704067260], "s"), + ([1704067200000, 1704067260000], "ms"), + ([1704067200000000, 1704067260000000], "us"), + ([1704067200000000000, 1704067260000000000], "ns"), + ], +) +def test_every_epoch_unit_inferred_and_explicit(values, unit): + df = pd.DataFrame({"ts": values}) + for event_time_unit in (None, unit): + rep = fd.cdc_profile( + df, event_time="ts", now="2024-01-01 00:05", event_time_unit=event_time_unit + ) + assert rep.freshness_seconds == pytest.approx(240.0) + + +def test_float_epoch_seconds_and_nullable_ints(): + df = pd.DataFrame({"ts": [1704067200.5, 1704067260.0]}) + rep = fd.cdc_profile(df, event_time="ts", now="2024-01-01 00:05") + assert rep.freshness_seconds == pytest.approx(240.0) + + df = pd.DataFrame({"ts": pd.array([1704067200000, None], dtype="Int64")}) + rep = fd.cdc_profile(df, event_time="ts", now="2024-01-01 00:05") + assert rep.freshness_seconds == pytest.approx(300.0) + assert _counts(rep) == {"missing_event_time": 1} + + +def test_explicit_event_time_unit_overrides_inference(): + # Millisecond magnitudes, but the caller says they are microseconds. + df = pd.DataFrame({"ts": [1704067200000]}) + rep = fd.cdc_profile(df, event_time="ts", now="1970-01-20 18:00", event_time_unit="us") + assert rep.freshness_seconds == pytest.approx( + (pd.Timestamp("1970-01-20 18:00") - pd.Timestamp(1704067200000, unit="us")).total_seconds() + ) + + +def test_event_time_unit_ignored_for_string_columns(): + df = pd.DataFrame({"ts": ["2024-01-01 00:01"]}) + rep = fd.cdc_profile(df, event_time="ts", now="2024-01-01 00:05", event_time_unit="ms") + assert rep.freshness_seconds == pytest.approx(240.0) + + +def test_invalid_event_time_unit_raises(): + df = pd.DataFrame({"ts": [1704067200000]}) + with pytest.raises(ValueError, match="event_time_unit"): + fd.cdc_profile(df, event_time="ts", event_time_unit="minutes") + + +# -- #328: replay_risk needs a duplicate key ------------------------------------ + + +def _late_only_frame() -> pd.DataFrame: + ts = ["2024-01-01 10:00"] + [f"2024-01-01 00:0{i}" for i in range(9)] + return pd.DataFrame({"ts": pd.to_datetime(ts), "k": ["a"] * 10}) + + +def test_late_only_batch_never_raises_replay_risk(): + rep = fd.cdc_profile( + _late_only_frame(), + event_time="ts", + key="k", + now="2024-01-02", + lateness="1m", + replay_threshold=0.2, + ) + assert _counts(rep) == {"late": 9} + + +def test_replay_risk_fires_once_a_duplicate_key_is_present(): + df = _late_only_frame() + df = pd.concat([df, df.iloc[[1]]], ignore_index=True) # duplicate one late row + rep = fd.cdc_profile( + df, event_time="ts", key="k", now="2024-01-02", lateness="1m", replay_threshold=0.2 + ) + counts = _counts(rep) + assert counts["duplicate_key"] == 2 + assert counts["replay_risk"] == counts["duplicate_key"] + counts["late"]