fix(cdc): null-key ordering checks, zero stale_after, epoch event times, replay_risk docs

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)}

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"]