ISSUE #26706 - Migrate Failed sample rows to OS

ingestion/src/metadata/data_quality/api/models.py

@@ -22,7 +22,7 @@
 
 from metadata.config.common import ConfigModel
 from metadata.generated.schema.api.tests.createTestSuite import CreateTestSuiteRequest
-from metadata.generated.schema.entity.data.table import Table
+from metadata.generated.schema.entity.data.table import Table, TableData
 from metadata.generated.schema.entity.services.databaseService import DatabaseConnection
 from metadata.generated.schema.tests.basic import TestCaseResult
 from metadata.generated.schema.tests.testCase import TestCase, TestCaseParameterValue
@@ -51,6 +51,9 @@ class TestSuiteProcessorConfig(ConfigModel):
 class TestCaseResultResponse(BaseModel):
     testCaseResult: TestCaseResult
     testCase: TestCase
+    failedRowsSample: Optional[TableData] = None
+    inspectionQuery: Optional[str] = None
+    validateColumns: bool = True
 
 
 class TableAndTests(BaseModel):

ingestion/src/metadata/data_quality/interface/test_suite_interface.py

@@ -17,6 +17,7 @@
 from abc import ABC, abstractmethod
 from typing import Optional, Set, Type
 
+from metadata.data_quality.api.models import TestCaseResultResponse
 from metadata.data_quality.builders.validator_builder import ValidatorBuilder
 from metadata.data_quality.validations.base_test_handler import BaseTestValidator
 from metadata.data_quality.validations.runtime_param_setter.param_setter import (
@@ -27,7 +28,7 @@
 )
 from metadata.generated.schema.entity.data.table import Table
 from metadata.generated.schema.entity.services.databaseService import DatabaseConnection
-from metadata.generated.schema.tests.basic import TestCaseResult, TestCaseStatus
+from metadata.generated.schema.tests.basic import TestCaseStatus
 from metadata.generated.schema.tests.testCase import TestCase
 from metadata.generated.schema.tests.testDefinition import TestDefinition
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -108,7 +109,7 @@ def _set_runtime_params_setter_fact(
         """
         cls.runtime_params_setter_fact = class_fact
 
-    def run_test_case(self, test_case: TestCase) -> Optional[TestCaseResult]:
+    def run_test_case(self, test_case: TestCase) -> Optional[TestCaseResultResponse]:
         """run column data quality tests"""
         runtime_params_setter_fact: RuntimeParameterSetterFactory = (
             self._get_runtime_params_setter_fact()
@@ -138,11 +139,14 @@ def run_test_case(self, test_case: TestCase) -> Optional[TestCaseResult]:
                 f"Error executing {test_case.testDefinition.fullyQualifiedName} - {err}"
             )
             logger.exception(message)
-            return validator.get_test_case_result_object(
-                validator.execution_date,
-                TestCaseStatus.Aborted,
-                message,
-                [],
+            return TestCaseResultResponse(
+                testCase=test_case,
+                testCaseResult=validator.get_test_case_result_object(
+                    validator.execution_date,
+                    TestCaseStatus.Aborted,
+                    message,
+                    [],
+                ),
             )
 
     def _get_table_config(self):

ingestion/src/metadata/data_quality/runner/core.py

@@ -14,7 +14,6 @@
 """
 
 
-from metadata.data_quality.api.models import TestCaseResultResponse
 from metadata.data_quality.interface.test_suite_interface import TestSuiteInterface
 from metadata.generated.schema.tests.testCase import TestCase
 from metadata.utils.logger import test_suite_logger
@@ -34,12 +33,8 @@ def run_and_handle(self, test_case: TestCase):
             f"Executing test case {test_case.name.root} "
             f"for entity {self.test_runner_interface.table_entity.fullyQualifiedName.root}"
         )
-        test_result = self.test_runner_interface.run_test_case(
+        result = self.test_runner_interface.run_test_case(
             test_case,
         )
 
-        if test_result:
-            return TestCaseResultResponse(
-                testCaseResult=test_result, testCase=test_case
-            )
-        return None
+        return result

ingestion/src/metadata/data_quality/validations/base_test_handler.py

@@ -32,6 +32,7 @@
 
 from pydantic import BaseModel
 
+from metadata.data_quality.api.models import TestCaseResultResponse
 from metadata.data_quality.validations import utils
 from metadata.data_quality.validations.impact_score import (
     DEFAULT_TOP_DIMENSIONS,
@@ -134,7 +135,7 @@ def _get_top_dimensions(self) -> int:
             return DEFAULT_TOP_DIMENSIONS
         return min(value, MAX_TOP_DIMENSIONS)
 
-    def run_validation(self) -> TestCaseResult:
+    def run_validation(self) -> TestCaseResultResponse:
         """Template method defining the validation flow with optional dimensional analysis
 
         This method orchestrates the overall validation process:
@@ -186,7 +187,20 @@ def run_validation(self) -> TestCaseResult:
                 )
                 logger.debug(traceback.format_exc())
 
-        return test_result
+        result: TestCaseResultResponse = TestCaseResultResponse(
+            testCaseResult=test_result, testCase=self.test_case
+        )
+
+        self.result_with_failed_samples(result)
+
+        return result
+
+    def result_with_failed_samples(self, result: TestCaseResultResponse) -> None:
+        """Hook for failed row sampling. No-op by default.
+
+        Overridden by FailedSampleValidatorMixin to fetch and stash
+        failed row samples on the validator instance.
+        """
 
     @abstractmethod
     def _run_validation(self) -> TestCaseResult:

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

@@ -28,10 +28,17 @@
     BaseColumnValueLengthsToBeBetweenValidator,
 )
 from metadata.data_quality.validations.impact_score import calculate_impact_score_pandas
+from metadata.data_quality.validations.mixins.failed_row_sampler_mixin import (
+    PandasFailedRowSamplerMixin,
+)
+from metadata.data_quality.validations.mixins.failed_sample_validator_mixin import (
+    FailedSampleValidatorMixin,
+)
 from metadata.data_quality.validations.mixins.pandas_validator_mixin import (
     PandasValidatorMixin,
     aggregate_others_statistical_pandas,
 )
+from metadata.generated.schema.entity.data.table import TableData
 from metadata.generated.schema.tests.dimensionResult import DimensionResult
 from metadata.profiler.metrics.registry import Metrics
 from metadata.utils.logger import test_suite_logger
@@ -41,7 +48,10 @@
 
 
 class ColumnValueLengthsToBeBetweenValidator(
-    BaseColumnValueLengthsToBeBetweenValidator, PandasValidatorMixin
+    FailedSampleValidatorMixin,
+    BaseColumnValueLengthsToBeBetweenValidator,
+    PandasValidatorMixin,
+    PandasFailedRowSamplerMixin,
 ):
     """Validator for column value lengths to be between test case"""
 
@@ -238,3 +248,21 @@ def compute_row_count(self, column: SQALikeColumn, min_bound: int, max_bound: in
         )
 
         return row_count, failed_rows
+
+    def filter(self):
+        min_bound = self.get_min_bound("minLength")
+        max_bound = self.get_max_bound("maxLength")
+        filters = []
+        if min_bound is not None and min_bound > float("-inf"):
+            filters.append(
+                f"{self.get_column().name}.astype('str').str.len() < {min_bound}"
+            )
+        if max_bound is not None and max_bound < float("inf"):
+            filters.append(
+                f"{self.get_column().name}.astype('str').str.len() > {max_bound}"
+            )
+        return " or ".join(filters)
+
+    def fetch_failed_rows_sample(self):
+        cols, rows = self._get_failed_rows_sample()
+        return TableData(columns=cols, rows=rows)

ingestion/src/metadata/data_quality/validations/column/pandas/columnValuesToBeBetween.py

@@ -14,6 +14,7 @@
 """
 
 from collections import defaultdict
+from datetime import datetime
 from typing import List, Optional, cast
 
 import pandas as pd
@@ -27,20 +28,32 @@
     BaseColumnValuesToBeBetweenValidator,
 )
 from metadata.data_quality.validations.impact_score import calculate_impact_score_pandas
+from metadata.data_quality.validations.mixins.failed_row_sampler_mixin import (
+    PandasFailedRowSamplerMixin,
+)
+from metadata.data_quality.validations.mixins.failed_sample_validator_mixin import (
+    FailedSampleValidatorMixin,
+)
 from metadata.data_quality.validations.mixins.pandas_validator_mixin import (
     PandasValidatorMixin,
     aggregate_others_statistical_pandas,
 )
+from metadata.generated.schema.entity.data.table import TableData
 from metadata.generated.schema.tests.dimensionResult import DimensionResult
 from metadata.profiler.metrics.registry import Metrics
+from metadata.profiler.orm.registry import is_date_time
 from metadata.utils.logger import test_suite_logger
 from metadata.utils.sqa_like_column import SQALikeColumn
+from metadata.utils.time_utils import convert_timestamp
 
 logger = test_suite_logger()
 
 
 class ColumnValuesToBeBetweenValidator(
-    BaseColumnValuesToBeBetweenValidator, PandasValidatorMixin
+    FailedSampleValidatorMixin,
+    BaseColumnValuesToBeBetweenValidator,
+    PandasValidatorMixin,
+    PandasFailedRowSamplerMixin,
 ):
     """Validator for column values to be between test case"""
 
@@ -237,3 +250,34 @@ def compute_row_count(self, column: SQALikeColumn, min_bound: int, max_bound: in
         )
 
         return row_count, failed_rows
+
+    def filter(self):
+        column = self.get_column()
+        if is_date_time(column.type):
+            min_bound = self.get_test_case_param_value(
+                self.test_case.parameterValues,
+                "minValue",
+                type_=datetime.fromtimestamp,
+                default=datetime.min,
+                pre_processor=convert_timestamp,
+            )
+            max_bound = self.get_test_case_param_value(
+                self.test_case.parameterValues,
+                "maxValue",
+                type_=datetime.fromtimestamp,
+                default=datetime.max,
+                pre_processor=convert_timestamp,
+            )
+        else:
+            min_bound = self.get_min_bound("minValue")
+            max_bound = self.get_max_bound("maxValue")
+        filters = []
+        if min_bound is not None:
+            filters.append(f"{column.name} < {min_bound}")
+        if max_bound is not None:
+            filters.append(f"{column.name} > {max_bound}")
+        return " or ".join(filters)
+
+    def fetch_failed_rows_sample(self):
+        cols, rows = self._get_failed_rows_sample()
+        return TableData(columns=cols, rows=rows)

ingestion/src/metadata/data_quality/validations/column/pandas/columnValuesToBeInSet.py

@@ -13,6 +13,7 @@
 Validator for column value to be in set test case
 """
 
+from ast import literal_eval
 from collections import defaultdict
 from typing import List, Optional, cast
 
@@ -27,10 +28,17 @@
     BaseColumnValuesToBeInSetValidator,
 )
 from metadata.data_quality.validations.impact_score import calculate_impact_score_pandas
+from metadata.data_quality.validations.mixins.failed_row_sampler_mixin import (
+    PandasFailedRowSamplerMixin,
+)
+from metadata.data_quality.validations.mixins.failed_sample_validator_mixin import (
+    FailedSampleValidatorMixin,
+)
 from metadata.data_quality.validations.mixins.pandas_validator_mixin import (
     PandasValidatorMixin,
     aggregate_others_pandas,
 )
+from metadata.generated.schema.entity.data.table import TableData
 from metadata.generated.schema.tests.dimensionResult import DimensionResult
 from metadata.profiler.metrics.core import add_props
 from metadata.profiler.metrics.registry import Metrics
@@ -41,7 +49,10 @@
 
 
 class ColumnValuesToBeInSetValidator(
-    BaseColumnValuesToBeInSetValidator, PandasValidatorMixin
+    FailedSampleValidatorMixin,
+    BaseColumnValuesToBeInSetValidator,
+    PandasValidatorMixin,
+    PandasFailedRowSamplerMixin,
 ):
     """Validator for column value to be in set test case"""
 
@@ -196,3 +207,15 @@ def compute_row_count(self, column: SQALikeColumn):
             NotImplementedError:
         """
         return self._compute_row_count(self.runner, column)
+
+    def filter(self):
+        items = self.get_test_case_param_value(
+            self.test_case.parameterValues,
+            "allowedValues",
+            literal_eval,
+        )
+        return f"~{self.get_column().name}.isin({items})"
+
+    def fetch_failed_rows_sample(self):
+        cols, rows = self._get_failed_rows_sample()
+        return TableData(columns=cols, rows=rows)

ingestion/src/metadata/data_quality/validations/column/pandas/columnValuesToBeNotInSet.py

@@ -13,6 +13,7 @@
 Validator for column value to be not in set test case
 """
 
+from ast import literal_eval
 from collections import defaultdict
 from typing import List, Optional, cast
 
@@ -27,10 +28,17 @@
     BaseColumnValuesToBeNotInSetValidator,
 )
 from metadata.data_quality.validations.impact_score import calculate_impact_score_pandas
+from metadata.data_quality.validations.mixins.failed_row_sampler_mixin import (
+    PandasFailedRowSamplerMixin,
+)
+from metadata.data_quality.validations.mixins.failed_sample_validator_mixin import (
+    FailedSampleValidatorMixin,
+)
 from metadata.data_quality.validations.mixins.pandas_validator_mixin import (
     PandasValidatorMixin,
     aggregate_others_pandas,
 )
+from metadata.generated.schema.entity.data.table import TableData
 from metadata.generated.schema.tests.dimensionResult import DimensionResult
 from metadata.profiler.metrics.core import add_props
 from metadata.profiler.metrics.registry import Metrics
@@ -41,7 +49,10 @@
 
 
 class ColumnValuesToBeNotInSetValidator(
-    BaseColumnValuesToBeNotInSetValidator, PandasValidatorMixin
+    FailedSampleValidatorMixin,
+    BaseColumnValuesToBeNotInSetValidator,
+    PandasValidatorMixin,
+    PandasFailedRowSamplerMixin,
 ):
     """Validator for column value to be not in set test case"""
 
@@ -182,3 +193,15 @@ def compute_row_count(self, column: SQALikeColumn):
             NotImplementedError:
         """
         return self._compute_row_count(self.runner, column)
+
+    def filter(self):
+        items = self.get_test_case_param_value(
+            self.test_case.parameterValues,
+            "forbiddenValues",
+            literal_eval,
+        )
+        return f"{self.get_column().name}.isin({items})"
+
+    def fetch_failed_rows_sample(self):
+        cols, rows = self._get_failed_rows_sample()
+        return TableData(columns=cols, rows=rows)