Feature/dimensionality column max to be between (open-metadata#24051)

* Initial implementation for Dimensionality on Data Quality Tests

* Fix ColumnValuesToBeUnique and create TestCaseResult API

* Refactor dimension result

* Initial E2E Implementation without Impact Score

* Dimensionality Thin Slice

* Update generated TypeScript types

* Update generated TypeScript types

* Removed useless method to use the one we already had

* Fix Pandas Dimensionality checks

* Remove useless comments

* Implement PR comments, fix Tests

* Improve the code a bit

* Fix imports

* Implement Dimensionality for ColumnMeanToBeBetween

* Removed useless comments and improved minor things

* Implement UnitTests

* Fixes

* Moved import pandas to type checking

* Fix Min/Max being optional

* Fix Unittests

* small fixes

* Fix Unittests

* Fix Issue with counting total rows on mean

* Improve code

* Fix Merge

* Removed unused type

* Fix Tests

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

Author: IceS2

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

@@ -78,9 +78,9 @@ class TestEvaluation(TypedDict, total=False):
     """
 
     matched: bool
-    passed_rows: int
-    failed_rows: int
-    total_rows: int
+    passed_rows: Optional[int]
+    failed_rows: Optional[int]
+    total_rows: Optional[int]
 
 
 class DimensionInfo(TypedDict):

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

@@ -15,11 +15,16 @@
 
 import traceback
 from abc import abstractmethod
-from typing import Union
+from typing import List, Optional, Union
 
 from sqlalchemy import Column
 
-from metadata.data_quality.validations.base_test_handler import BaseTestValidator
+from metadata.data_quality.validations.base_test_handler import (
+    BaseTestValidator,
+    DimensionInfo,
+    DimensionResult,
+    TestEvaluation,
+)
 from metadata.generated.schema.tests.basic import (
     TestCaseResult,
     TestCaseStatus,
@@ -31,8 +36,6 @@
 
 logger = test_suite_logger()
 
-MAX = "max"
-
 
 class BaseColumnValueMaxToBeBetweenValidator(BaseTestValidator):
     """Validator for column value max to be between test case"""
@@ -46,9 +49,16 @@ def _run_validation(self) -> TestCaseResult:
         Returns:
             TestCaseResult: The test case result for the overall validation
         """
+        test_params = self._get_test_parameters()
+
         try:
             column: Union[SQALikeColumn, Column] = self._get_column_name()
-            res = self._run_results(Metrics.MAX, column)
+            max_value = self._run_results(Metrics.MAX, column)
+
+            metric_values = {
+                Metrics.MAX.name: max_value,
+            }
+
         except (ValueError, RuntimeError) as exc:
             msg = f"Error computing {self.test_case.fullyQualifiedName}: {exc}"  # type: ignore
             logger.debug(traceback.format_exc())
@@ -57,25 +67,210 @@ def _run_validation(self) -> TestCaseResult:
                 self.execution_date,
                 TestCaseStatus.Aborted,
                 msg,
-                [TestResultValue(name=MAX, value=None)],
+                [TestResultValue(name=Metrics.MAX.name, value=None)],
             )
 
-        min_bound = self.get_min_bound("minValueForMaxInCol")
-        max_bound = self.get_max_bound("maxValueForMaxInCol")
+        evaluation = self._evaluate_test_condition(metric_values, test_params)
+        result_message = self._format_result_message(
+            metric_values, test_params=test_params
+        )
+        test_result_values = self._get_test_result_values(metric_values)
 
         return self.get_test_case_result_object(
             self.execution_date,
-            self.get_test_case_status(min_bound <= res <= max_bound),
-            f"Found max={res} vs.  the expected min={min_bound}, max={max_bound}.",
-            [TestResultValue(name=MAX, value=str(res))],
-            min_bound=min_bound,
-            max_bound=max_bound,
+            self.get_test_case_status(evaluation["matched"]),
+            result_message,
+            test_result_values,
+            min_bound=test_params["minValueForMaxInCol"],
+            max_bound=test_params["maxValueForMaxInCol"],
         )
 
+    def _run_dimensional_validation(self) -> List[DimensionResult]:
+        """Execute dimensional validation for column value max to be between
+
+        The new approach runs separate queries for each dimension column instead of
+        combining them with GROUP BY. For example, if dimensionColumns = ["region", "category"],
+        this method will:
+        1. Run one query: GROUP BY region -> {"North America": result1, "Europe": result2}
+        2. Run another query: GROUP BY category -> {"Electronics": result3, "Clothing": result4}
+
+        Returns:
+            List[DimensionResult]: List of dimension-specific test results
+        """
+        try:
+            dimension_columns = self.test_case.dimensionColumns or []
+            if not dimension_columns:
+                return []
+
+            column: Union[SQALikeColumn, Column] = self._get_column_name()
+
+            test_params = self._get_test_parameters()
+            metrics_to_compute = self._get_metrics_to_compute(test_params)
+
+            dimension_results = []
+            for dimension_column in dimension_columns:
+                try:
+                    dimension_col = self._get_column_name(dimension_column)
+
+                    single_dimension_results = self._execute_dimensional_validation(
+                        column, dimension_col, metrics_to_compute, test_params
+                    )
+
+                    dimension_results.extend(single_dimension_results)
+
+                except Exception as exc:
+                    logger.warning(
+                        f"Error executing dimensional query for column {dimension_column}: {exc}"
+                    )
+                    logger.debug(traceback.format_exc())
+                    continue
+
+            return dimension_results
+
+        except Exception as exc:
+            logger.warning(f"Error executing dimensional validation: {exc}")
+            logger.debug(traceback.format_exc())
+            return []
+
+    def _get_test_parameters(self) -> dict:
+        """Get test parameters for this validator
+
+        Returns:
+            dict: Test parameters including min and max bounds
+        """
+        return {
+            "minValueForMaxInCol": self.get_min_bound("minValueForMaxInCol"),
+            "maxValueForMaxInCol": self.get_max_bound("maxValueForMaxInCol"),
+        }
+
+    def _get_metrics_to_compute(self, test_params: Optional[dict] = None) -> dict:
+        """Get metrics that need to be computed for this test
+
+        Args:
+            test_params: Optional test parameters (unused for max validator)
+
+        Returns:
+            dict: Dictionary mapping metric names to Metrics enum values
+        """
+        return {
+            Metrics.MAX.name: Metrics.MAX,
+        }
+
+    def _evaluate_test_condition(
+        self, metric_values: dict, test_params: dict
+    ) -> TestEvaluation:
+        """Evaluate the max-to-be-between test condition
+
+        For max test, the condition passes if the max value is within the specified bounds.
+        Since this is a statistical validator (group-level), passed/failed row counts are not applicable.
+
+        Args:
+            metric_values: Dictionary with keys from Metrics enum names
+                          e.g., {"MAX": 42.5}
+            test_params: Dictionary with 'minValueForMaxInCol' and 'maxValueForMaxInCol'
+
+        Returns:
+            dict with keys:
+                - matched: bool - whether test passed (max within bounds)
+                - passed_rows: None - not applicable for statistical validators
+                - failed_rows: None - not applicable for statistical validators
+                - total_rows: None - not applicable for statistical validators
+        """
+        max_value = metric_values[Metrics.MAX.name]
+        min_bound = test_params["minValueForMaxInCol"]
+        max_bound = test_params["maxValueForMaxInCol"]
+
+        matched = min_bound <= max_value <= max_bound
+
+        return {
+            "matched": matched,
+            "passed_rows": None,
+            "failed_rows": None,
+            "total_rows": None,
+        }
+
+    def _format_result_message(
+        self,
+        metric_values: dict,
+        dimension_info: Optional[DimensionInfo] = None,
+        test_params: Optional[dict] = None,
+    ) -> str:
+        """Format the result message for max-to-be-between test
+
+        Args:
+            metric_values: Dictionary with Metrics enum names as keys
+            dimension_info: Optional DimensionInfo with dimension details
+            test_params: Test parameters with min/max bounds. Required for this test.
+
+        Returns:
+            str: Formatted result message
+        """
+        if test_params is None:
+            raise ValueError(
+                "test_params is required for columnValueMaxToBeBetween._format_result_message"
+            )
+
+        max_value = metric_values[Metrics.MAX.name]
+        min_bound = test_params["minValueForMaxInCol"]
+        max_bound = test_params["maxValueForMaxInCol"]
+
+        if dimension_info:
+            return (
+                f"Dimension {dimension_info['dimension_name']}={dimension_info['dimension_value']}: "
+                f"Found max={max_value} vs. the expected min={min_bound}, max={max_bound}"
+            )
+        else:
+            return f"Found max={max_value} vs. the expected min={min_bound}, max={max_bound}."
+
+    def _get_test_result_values(self, metric_values: dict) -> List[TestResultValue]:
+        """Get test result values for max-to-be-between test
+
+        Args:
+            metric_values: Dictionary with Metrics enum names as keys
+
+        Returns:
+            List[TestResultValue]: Test result values for the test case
+        """
+        return [
+            TestResultValue(
+                name=Metrics.MAX.name,
+                value=str(metric_values[Metrics.MAX.name]),
+            ),
+        ]
+
     @abstractmethod
-    def _get_column_name(self):
+    def _get_column_name(self, column_name: Optional[str] = None):
+        """Get column object from entity link or column name
+
+        Args:
+            column_name: Optional column name. If None, returns the test case column.
+
+        Returns:
+            Column object
+        """
         raise NotImplementedError
 
     @abstractmethod
     def _run_results(self, metric: Metrics, column: Union[SQALikeColumn, Column]):
         raise NotImplementedError
+
+    @abstractmethod
+    def _execute_dimensional_validation(
+        self,
+        column: Union[SQALikeColumn, Column],
+        dimension_col: Union[SQALikeColumn, Column],
+        metrics_to_compute: dict,
+        test_params: dict,
+    ) -> List[DimensionResult]:
+        """Execute dimensional validation query for a single dimension column
+
+        Args:
+            column: The column being tested (e.g., revenue)
+            dimension_col: The dimension column to group by (e.g., region)
+            metrics_to_compute: Dict mapping metric names to Metrics enum values
+            test_params: Test parameters including min and max bounds
+
+        Returns:
+            List of DimensionResult objects for each dimension value
+        """
+        raise NotImplementedError