feat(detector): compare distinct ratio for historically unique columns (#94)

src/scherlok/detector/cardinality.py

@@ -9,18 +9,44 @@
 CARDINALITY_WARNING_PCT = 50   # 50% change
 CARDINALITY_CRITICAL_PCT = 200  # 3x change (e.g. status went from 5 to 500)
 
+# Ratio threshold for historically near-unique columns.
+# When stored distinct_count / row_count >= UNIQUE_RATIO the column is
+# considered near-unique and we compare distinct ratios instead of absolute
+# distinct counts, so a volume drop does not fire cardinality noise.
+UNIQUE_RATIO = 0.95
+# Ratio changes are bounded to 0-100%, so the absolute cardinality thresholds
+# above cannot classify a near-unique column as critical.
+UNIQUE_RATIO_WARNING_PCT = 10
+UNIQUE_RATIO_CRITICAL_PCT = 50
+# Tolerance for binary floating-point boundaries (e.g. 1 - 0.9 is
+# 0.09999999999999998, so a clean 10% loss would fall just under the warning
+# band without this).
+_RATIO_EPS = 1e-9
+
 
 def detect_cardinality_anomalies(
     table: str,
     column: str,
     current_dist: dict,
     stored_dist: dict,
+    current_vol: dict | None = None,
+    stored_vol: dict | None = None,
     *,
     history: Sequence[dict] | None = None,
 ) -> list[dict]:
     """Compare current distinct count against stored profile for a column.
 
     Returns anomalies when cardinality changes significantly.
+
+    When the stored profile shows the column was near-unique
+    (distinct_count / non-null rows >= UNIQUE_RATIO) the distinct ratio
+    (distinct / non-null rows) is compared instead of the absolute count. A
+    unique column that stays unique after a volume change is then silent,
+    while a unique column that suddenly has duplicates still fires.
+    The denominator excludes NULLs (rows - null_count from each
+    distribution profile) so nulling values in a unique column does not
+    read as a uniqueness loss. Falls back to absolute comparison when
+    volume profiles are missing or non-null row counts are zero.
     """
     anomalies: list[dict] = []
 
@@ -63,6 +89,64 @@ def detect_cardinality_anomalies(
     if stored_card == 0:
         return anomalies
 
+    # Near-unique path: use distinct ratio when stored was near-unique
+    # and both volume profiles provide valid row counts. The denominator is
+    # non-null rows (rows - null_count) so NULLing values in a unique column
+    # does not read as a uniqueness loss.
+    if current_vol is not None and stored_vol is not None:
+        stored_rows = stored_vol.get("row_count")
+        current_rows = current_vol.get("row_count")
+        if (
+            isinstance(stored_rows, int)
+            and isinstance(current_rows, int)
+            and stored_rows > 0
+            and current_rows > 0
+            and isinstance(stored_card, int)
+            and isinstance(current_card, int)
+        ):
+            stored_nulls = stored_dist.get("null_count", 0) or 0
+            current_nulls = current_dist.get("null_count", 0) or 0
+            if not isinstance(stored_nulls, int):
+                stored_nulls = 0
+            if not isinstance(current_nulls, int):
+                current_nulls = 0
+            stored_denom = stored_rows - stored_nulls
+            current_denom = current_rows - current_nulls
+            if stored_denom <= 0 or current_denom <= 0:
+                pass
+            else:
+                stored_ratio = stored_card / stored_denom
+                if stored_ratio >= UNIQUE_RATIO:
+                    current_ratio = current_card / current_denom
+                    change_pct = abs(current_ratio - stored_ratio) / stored_ratio * 100
+                    direction = "decreased" if current_ratio < stored_ratio else "increased"
+                    # Inclusive on both bands to match the absolute path below;
+                    # _RATIO_EPS keeps exact decimal boundaries (10%, 50%)
+                    # from falling just under on binary floats.
+                    if change_pct + _RATIO_EPS >= UNIQUE_RATIO_CRITICAL_PCT:
+                        anomalies.append({
+                            "table": table,
+                            "type": "cardinality_change",
+                            "message": (
+                                f"Column '{column}' distinct ratio {direction}: "
+                                f"{stored_ratio:.3f} -> {current_ratio:.3f} "
+                                f"({change_pct:.0f}% change)"
+                            ),
+                            "severity": Severity.CRITICAL,
+                        })
+                    elif change_pct + _RATIO_EPS >= UNIQUE_RATIO_WARNING_PCT:
+                        anomalies.append({
+                            "table": table,
+                            "type": "cardinality_change",
+                            "message": (
+                                f"Column '{column}' distinct ratio {direction}: "
+                                f"{stored_ratio:.3f} -> {current_ratio:.3f} "
+                                f"({change_pct:.0f}% change)"
+                            ),
+                            "severity": Severity.WARNING,
+                        })
+                    return anomalies
+
     change_pct = abs(current_card - stored_card) / stored_card * 100
     direction = "increased" if current_card > stored_card else "decreased"
 

src/scherlok/service.py

@@ -84,7 +84,13 @@ def profile_and_detect(
             )
             anomalies.extend(
                 detect_cardinality_anomalies(
-                    table, col_name, current_dist, stored_dist, history=distribution_history
+                    table,
+                    col_name,
+                    current_dist,
+                    stored_dist,
+                    current_vol,
+                    stored_vol,
+                    history=distribution_history,
                 )
             )
         store.save_profile(table, f"distribution:{col_name}", current_dist)

tests/test_detector.py

@@ -300,3 +300,152 @@ def test_no_anomaly_zero_stored(self):
         stored = {"distinct_count": 0}
         anomalies = detect_cardinality_anomalies("t", "col", current, stored)
         assert len(anomalies) == 0
+
+
+class TestCardinalityNearUniqueColumns:
+    """Use distinct ratio for columns that were historically near-unique."""
+
+    STORED_DIST = {"distinct_count": 1000}
+    STORED_VOL = {"row_count": 1000}
+
+    def test_volume_drop_on_unique_column_is_silent(self):
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 400},
+            self.STORED_DIST,
+            {"row_count": 400},
+            self.STORED_VOL,
+        )
+        assert anomalies == []
+
+    def test_duplicates_on_unique_column_still_fire(self):
+        # An exact 50% uniqueness loss is CRITICAL (>= on both bands,
+        # matching the absolute path).
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 500},
+            self.STORED_DIST,
+            {"row_count": 1000},
+            self.STORED_VOL,
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.CRITICAL
+        assert "decreased" in anomalies[0]["message"]
+        assert "distinct ratio" in anomalies[0]["message"]
+
+    def test_exact_warning_boundary_fires_despite_float_repr(self):
+        # 1 - 0.9 is 0.09999999999999998 in binary floats; a clean 10%
+        # loss must still warn.
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 900},
+            self.STORED_DIST,
+            {"row_count": 1000},
+            self.STORED_VOL,
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.WARNING
+        assert "distinct ratio" in anomalies[0]["message"]
+
+    def test_partial_loss_below_critical_stays_warning(self):
+        # 1000 -> 800 is a 20% uniqueness loss: clearly warning, and clear
+        # of the 10% float edge.
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 800},
+            self.STORED_DIST,
+            {"row_count": 1000},
+            self.STORED_VOL,
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.WARNING
+        assert "distinct ratio" in anomalies[0]["message"]
+
+    def test_nulling_unique_values_is_silent(self):
+        # 20% of a unique e-mail column nulled: 1,600 remaining values are
+        # still distinct, so the non-null ratio is unchanged and nothing fires.
+        anomalies = detect_cardinality_anomalies(
+            "users",
+            "email",
+            {"distinct_count": 1600, "null_count": 400},
+            {"distinct_count": 2000, "null_count": 0},
+            {"row_count": 2000},
+            {"row_count": 2000},
+        )
+        assert anomalies == []
+
+    def test_null_aware_ratio_still_fires_on_real_duplicates(self):
+        # Same 20% NULLs, but only 800 distinct among the 1,600 non-null:
+        # uniqueness halved, so it must fire.
+        anomalies = detect_cardinality_anomalies(
+            "users",
+            "email",
+            {"distinct_count": 800, "null_count": 400},
+            {"distinct_count": 2000, "null_count": 0},
+            {"row_count": 2000},
+            {"row_count": 2000},
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.CRITICAL
+        assert "distinct ratio" in anomalies[0]["message"]
+
+    def test_low_cardinality_column_keeps_absolute_comparison(self):
+        anomalies = detect_cardinality_anomalies(
+            "users",
+            "plan",
+            {"distinct_count": 605},
+            {"distinct_count": 5},
+            {"row_count": 400},
+            {"row_count": 1000},
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.CRITICAL
+        assert "distinct values" in anomalies[0]["message"]
+
+    def test_falls_back_to_absolute_without_volume_profiles(self):
+        anomalies = detect_cardinality_anomalies(
+            "orders", "id", {"distinct_count": 400}, self.STORED_DIST
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.WARNING
+
+    def test_falls_back_to_absolute_on_zero_row_counts(self):
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 400},
+            self.STORED_DIST,
+            {"row_count": 0},
+            {"row_count": 0},
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.WARNING
+
+    def test_column_just_below_unique_threshold_uses_absolute(self):
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "customer_id",
+            {"distinct_count": 360},
+            {"distinct_count": 900},
+            {"row_count": 400},
+            {"row_count": 1000},
+        )
+        assert len(anomalies) == 1
+        assert "distinct values" in anomalies[0]["message"]
+
+    def test_large_loss_of_uniqueness_is_critical(self):
+        anomalies = detect_cardinality_anomalies(
+            "orders",
+            "id",
+            {"distinct_count": 100},
+            self.STORED_DIST,
+            {"row_count": 1000},
+            self.STORED_VOL,
+        )
+        assert len(anomalies) == 1
+        assert anomalies[0]["severity"] == Severity.CRITICAL
+        assert "distinct ratio" in anomalies[0]["message"]