Fixes #33231: [DQ Threshold W2.4] Build row-level violation counts for columnValuesToBeBetween / LengthsToBeBetween

.github/playwright/impact-map.json

@@ -12,7 +12,7 @@
     {
       "projects": ["Basic"],
       "specs": [
-        "playwright/e2e/Pages/Login.spec.ts",
+        "playwright/e2e/Auth/Login.spec.ts",
         "playwright/e2e/Flow/Navbar.spec.ts",
         "playwright/e2e/Features/Dashboards.spec.ts",
         "playwright/e2e/Pages/Lineage/LineageControls.spec.ts",
@@ -354,7 +354,7 @@
       "specs": [
         "playwright/e2e/Features/SettingsNavigationPage.spec.ts",
         "playwright/e2e/Flow/LineageSettings.spec.ts",
-        "playwright/e2e/Pages/LoginConfiguration.spec.ts",
+        "playwright/e2e/Auth/LoginConfiguration.spec.ts",
         "playwright/e2e/Pages/OmdURLConfiguration.spec.ts",
         "playwright/e2e/Pages/ProfilerConfigurationPage.spec.ts",
         "playwright/e2e/Pages/SearchSettings.spec.ts",
@@ -393,7 +393,7 @@
       "projects": ["chromium", "Basic", "Ingestion", "SearchRBAC", "ImportExport"],
       "specs": [
         "playwright/e2e/**/*Permission*.spec.ts",
-        "playwright/e2e/Pages/Login*.spec.ts",
+        "playwright/e2e/Auth/Login*.spec.ts",
         "playwright/e2e/Flow/SearchRBAC.spec.ts"
       ]
     },
@@ -467,7 +467,7 @@
       ],
       "projects": ["chromium", "Basic"],
       "specs": [
-        "playwright/e2e/Pages/Login.spec.ts",
+        "playwright/e2e/Auth/Login.spec.ts",
         "playwright/e2e/Flow/Navbar.spec.ts"
       ]
     },

ingestion/src/metadata/data_quality/validations/checkers/between_bounds_checker.py

@@ -22,7 +22,7 @@
 )
 
 if TYPE_CHECKING:
-    from sqlalchemy.sql.elements import ClauseElement
+    from sqlalchemy.sql.elements import ClauseElement, ColumnElement
 
 
 class BetweenBoundsChecker(BaseValidationChecker):
@@ -32,6 +32,17 @@ def __init__(self, min_bound: float, max_bound: float):
         self.min_bound = min_bound
         self.max_bound = max_bound
 
+    def _is_unbounded(self, bound: Any) -> bool:
+        """Whether that side of the window lets everything through
+
+        An unset bound resolves to ∓inf, and to None for a validator that resolves its bounds
+        dynamically and found none on that side. Neither excludes a value, so neither needs a
+        condition -- and None cannot be compared against at all. Only a float can be infinite:
+        a datetime window -- what a between test on a date column resolves to -- compares fine
+        and must not be handed to `math.isinf`, which only takes a number.
+        """
+        return bound is None or (isinstance(bound, float) and math.isinf(bound))
+
     def _check_violations(self, values):
         """Core violation check logic - works for both scalar and Series.
 
@@ -46,7 +57,19 @@ def _check_violations(self, values):
         """
         import pandas as pd
 
-        return ~pd.isna(values) & ((values < self.min_bound) | (values > self.max_bound))
+        # Only the sides the window actually sets are compared against, like the SQL half
+        # builds only those conditions: an unset bound excludes nothing, and a None one cannot
+        # be compared against at all.
+        outside = None
+        if not self._is_unbounded(self.min_bound):
+            outside = values < self.min_bound
+        if not self._is_unbounded(self.max_bound):
+            above = values > self.max_bound
+            outside = above if outside is None else (outside | above)
+
+        # `& False` over a window with neither side set keeps the shape of `values`, so a caller
+        # masking a Series with the result still gets a Series back.
+        return ~pd.isna(values) & (False if outside is None else outside)
 
     def _value_violates(self, value: Any) -> bool:
         """Check violation of one value (scalar).
@@ -85,9 +108,9 @@ def build_violation_sqa(self, metrics: list["ClauseElement"]) -> "ClauseElement"
         for expr in metrics:
             expr_conditions = []
 
-            if not math.isinf(self.min_bound):
+            if not self._is_unbounded(self.min_bound):
                 expr_conditions.append(and_(expr.isnot(None), expr < self.min_bound))
-            if not math.isinf(self.max_bound):
+            if not self._is_unbounded(self.max_bound):
                 expr_conditions.append(and_(expr.isnot(None), expr > self.max_bound))
 
             if expr_conditions:
@@ -96,7 +119,7 @@ def build_violation_sqa(self, metrics: list["ClauseElement"]) -> "ClauseElement"
             return literal(False)
         return or_(*conditions) if len(conditions) > 1 else conditions[0]
 
-    def build_row_level_violations_sqa(self, column: "ClauseElement") -> "ClauseElement":
+    def build_row_level_violations_sqa(self, column: "ClauseElement") -> "ColumnElement":
         """Build SQL expression to count row-level violations.
 
         Returns a SUM(CASE...) expression that counts individual rows where
@@ -115,15 +138,16 @@ def build_row_level_violations_sqa(self, column: "ClauseElement") -> "ClauseElem
         # Build condition: value NOT NULL AND (value < min OR value > max)
         conditions = []
 
-        if not math.isinf(self.min_bound):
+        if not self._is_unbounded(self.min_bound):
             conditions.append(and_(column.isnot(None), column < self.min_bound))
-        if not math.isinf(self.max_bound):
+        if not self._is_unbounded(self.max_bound):
             conditions.append(and_(column.isnot(None), column > self.max_bound))
 
         if not conditions:
             return literal(0)
 
         violation_condition = or_(*conditions) if len(conditions) > 1 else conditions[0]
 
-        # Return SUM(CASE WHEN violation THEN 1 ELSE 0 END)
-        return func.sum(case((violation_condition, literal(1)), else_=literal(0)))
+        # Return SUM(CASE WHEN violation THEN 1 ELSE 0 END). SUM over no row at all is NULL,
+        # which is not a count: a table with nothing in it has zero violations.
+        return func.coalesce(func.sum(case((violation_condition, literal(1)), else_=literal(0))), literal(0))

ingestion/src/metadata/data_quality/validations/column/base/columnValueLengthsToBeBetween.py

@@ -75,6 +75,14 @@ def _run_validation(self) -> TestCaseResult:
                 Metrics.minLength.name: min_res,
             }
 
+            if self._needs_violation_count():
+                # The counts go under the same two keys a dimension row carries its own under:
+                # the verdict and the message are read off `metric_values` by code shared with
+                # the dimensional path, which only finds them by those names.
+                total_rows, violating_rows = self._run_violation_count(column, test_params)
+                metric_values[DIMENSION_TOTAL_COUNT_KEY] = total_rows
+                metric_values[DIMENSION_FAILED_COUNT_KEY] = violating_rows
+
         except (ValueError, RuntimeError) as exc:
             msg = f"Error computing {self.test_case.fullyQualifiedName}: {exc}"  # type: ignore
             logger.debug(traceback.format_exc())
@@ -89,12 +97,16 @@ def _run_validation(self) -> TestCaseResult:
                 ],
             )
 
-        if self.test_case.computePassedFailedRowCount:
+        # A row tolerance already counted both, so report the counts the verdict was taken on
+        # rather than counting twice: a second scan reads the table again -- or, on a percentage
+        # sample, a different set of rows -- and could report rows that contradict the status.
+        row_count = metric_values.get(DIMENSION_TOTAL_COUNT_KEY)
+        failed_rows = metric_values.get(DIMENSION_FAILED_COUNT_KEY)
+
+        if failed_rows is None and self.test_case.computePassedFailedRowCount:
             row_count, failed_rows = self.compute_row_count(
                 column, test_params[self.MIN_BOUND], test_params[self.MAX_BOUND]
             )
-        else:
-            row_count, failed_rows = None, None
 
         evaluation = self._evaluate_test_condition(metric_values, test_params)
         result_message = self._format_result_message(metric_values, test_params=test_params)
@@ -120,18 +132,27 @@ def _get_validation_checker(self, test_params: dict) -> BetweenBoundsChecker:
     def _get_test_parameters(self) -> dict:
         """Get test parameters for this validator
 
+        The window is left exactly as the test case configured it. This test reads every row,
+        so its failure threshold is a row tolerance -- it is spent on how many values may fall
+        outside the length window, not on widening the window itself. Widening here as well
+        would apply the same tolerance twice.
+
         Returns:
-            dict: Test parameters including min and max bounds, widened by the failure threshold
+            dict: Test parameters including min and max bounds
         """
-        min_bound, max_bound = self.get_bounds(self.MIN_BOUND, self.MAX_BOUND)
         return {
-            self.MIN_BOUND: min_bound,
-            self.MAX_BOUND: max_bound,
+            self.MIN_BOUND: self.get_min_bound(self.MIN_BOUND),
+            self.MAX_BOUND: self.get_max_bound(self.MAX_BOUND),
         }
 
     def _get_metrics_to_compute(self, test_params: dict | None = None) -> dict:
         """Get metrics that need to be computed for this test
 
+        The values whose length falls outside the window are not a registry metric -- there is
+        no aggregate that answers "how many values are too short or too long". They are built
+        from the bounds by `BetweenBoundsChecker` instead, in `_run_violation_count()` for the
+        overall result and in `_execute_dimensional_validation()` for each dimension row.
+
         Args:
             test_params: Optional test parameters (unused for max validator)
 
@@ -143,11 +164,30 @@ def _get_metrics_to_compute(self, test_params: dict | None = None) -> dict:
             Metrics.minLength.name: Metrics.minLength,
         }
 
+    def _needs_violation_count(self) -> bool:
+        """Whether the values outside the length window have to be counted
+
+        Only a configured tolerance needs the count: with no tolerance, one value outside the
+        window and a shortest or longest value outside it are the same verdict, and counting
+        would cost a query nobody reads.
+        """
+        return bool(self.get_failure_threshold().value)
+
+    def _has_violation_count(self, metric_values: dict) -> bool:
+        """Whether this result was decided by counting rows rather than by the two extremes
+
+        The dimensional query counts violations for every test case, tolerance or not, so the
+        count alone does not mean a row tolerance applies. Without one the two agree anyway.
+        """
+        return self._needs_violation_count() and metric_values.get(DIMENSION_FAILED_COUNT_KEY) is not None
+
     def _evaluate_test_condition(self, metric_values: dict, test_params: dict) -> TestEvaluation:
         """Evaluate the max-to-be-between test condition
 
-        For dimensional validation, computes row-level passed/failed counts.
-        For non-dimensional validation, row counts are not applicable.
+        Without a tolerance the verdict is read off the two extremes: the shortest and the
+        longest value inside the window means every length is. That cannot answer "how many
+        rows are out of range", which is what a row tolerance is checked against, so a test
+        case that configures one is decided on the counted violations instead.
 
         Args:
             metric_values: Dictionary with keys from Metrics enum names
@@ -169,11 +209,15 @@ def _evaluate_test_condition(self, metric_values: dict, test_params: dict) -> Te
         min_bound = test_params[self.MIN_BOUND]
         max_bound = test_params[self.MAX_BOUND]
 
-        matched = min_bound <= min_length_value and max_length_value <= max_bound
-
-        # Extract row counts if available (dimensional validation)
+        # Extract row counts if available (row tolerance or dimensional validation)
         total_rows = metric_values.get(DIMENSION_TOTAL_COUNT_KEY)
         failed_rows = metric_values.get(DIMENSION_FAILED_COUNT_KEY)
+
+        if self._has_violation_count(metric_values):
+            matched = self._apply_row_threshold(failed_rows, total_rows)
+        else:
+            matched = min_bound <= min_length_value and max_length_value <= max_bound
+
         passed_rows = None
         if total_rows is not None and failed_rows is not None:
             passed_rows = total_rows - failed_rows
@@ -211,6 +255,22 @@ def _format_result_message(
 
         column = self.column_label()
 
+        if self._has_violation_count(metric_values):
+            # A row tolerance is a verdict on rows, so the message leads with the count it was
+            # checked against and keeps the window and its extremes as context.
+            counted = self.format_violation_message(
+                violations=metric_values.get(DIMENSION_FAILED_COUNT_KEY),
+                population=metric_values.get(DIMENSION_TOTAL_COUNT_KEY),
+                violation_noun=f"values in {column} outside the expected length",
+                matched=self._matched(metric_values, test_params),
+                dimension_info=dimension_info,
+            )
+            return (
+                f"{counted} Expected lengths {result_messages.bounds_phrase(min_bound, max_bound)}, "
+                f"with a shortest value of {result_messages.format_value(min_length_value)} characters "
+                f"and a longest of {result_messages.format_value(max_length_value)}."
+            )
+
         # Both extremes are checked against the same window, so the message reports both and
         # states the verdict once, on the pair.
         return self.format_statistic_message(
@@ -245,6 +305,23 @@ def _get_test_result_values(self, metric_values: dict) -> list[TestResultValue]:
     def _run_results(self, metric: Metrics, column: SQALikeColumn | Column):
         raise NotImplementedError
 
+    @abstractmethod
+    def _run_violation_count(self, column: SQALikeColumn | Column, test_params: dict) -> tuple[int | None, int | None]:
+        """Count the rows read and the ones whose length falls outside the window
+
+        Both halves are built by `BetweenBoundsChecker` so the SQL and the pandas engines
+        count the same rows: a NULL has no length and is not a violation, and an unset bound
+        excludes nothing.
+
+        Args:
+            column: the column under test
+            test_params: test parameters including min and max bounds
+
+        Returns:
+            tuple[int | None, int | None]: rows evaluated, rows outside the window
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def _execute_dimensional_validation(
         self,