[BUGFIX] Exclude nulls and fix the rank offset in the SQLite quantile metric

great_expectations/expectations/metrics/column_aggregate_metrics/column_quantile_values.py

@@ -3,18 +3,22 @@
 import ast
 import itertools
 import logging
+import math
 import traceback
 from collections.abc import Iterable
-from typing import Any
+from fractions import Fraction
+from typing import TYPE_CHECKING, Any, Optional
 
 import numpy as np
 
 from great_expectations.compatibility import sqlalchemy, trino
 from great_expectations.compatibility.sqlalchemy import (
     sqlalchemy as sa,
 )
+from great_expectations.compatibility.typing_extensions import override
 from great_expectations.core.metric_domain_types import MetricDomainTypes
 from great_expectations.execution_engine import (
+    ExecutionEngine,
     PandasExecutionEngine,
     SparkDFExecutionEngine,
     SqlAlchemyExecutionEngine,
@@ -27,6 +31,12 @@
 )
 from great_expectations.expectations.metrics.metric_provider import metric_value
 from great_expectations.expectations.metrics.util import attempt_allowing_relative_error
+from great_expectations.validator.metric_configuration import MetricConfiguration
+
+if TYPE_CHECKING:
+    from great_expectations.expectations.expectation_configuration import (
+        ExpectationConfiguration,
+    )
 
 logger = logging.getLogger(__name__)
 
@@ -72,7 +82,6 @@ def _sqlalchemy(  # noqa: C901, PLR0911 # FIXME CoP
         dialect_name = execution_engine.dialect_name
         quantiles = metric_value_kwargs["quantiles"]
         allow_relative_error = metric_value_kwargs.get("allow_relative_error", False)
-        table_row_count = metrics.get("table.row_count")
         if dialect_name == GXSqlDialect.SQL_SERVER:
             return _get_column_quantiles_sql_server(
                 column=column,
@@ -125,12 +134,17 @@ def _sqlalchemy(  # noqa: C901, PLR0911 # FIXME CoP
                 execution_engine=execution_engine,
             )
         elif dialect_name == GXSqlDialect.SQLITE:
+            # Subscript rather than "get", so a missing dependency raises instead of being read as
+            # an empty column and silently returning NaN.
+            nonnull_count = metrics["column_values.nonnull.count"]
+            if not nonnull_count:
+                return [np.nan] * len(quantiles)
             return _get_column_quantiles_sqlite(
                 column=column,
                 quantiles=quantiles,
                 selectable=selectable,
                 execution_engine=execution_engine,
-                table_row_count=table_row_count,
+                nonnull_count=nonnull_count,
             )
         elif dialect_name == GXSqlDialect.AWSATHENA:
             return _get_column_quantiles_athena(
@@ -191,6 +205,35 @@ def _spark(
 
         return quantile_values
 
+    @classmethod
+    @override
+    def _get_evaluation_dependencies(
+        cls,
+        metric: MetricConfiguration,
+        configuration: Optional[ExpectationConfiguration] = None,
+        execution_engine: Optional[ExecutionEngine] = None,
+        runtime_configuration: Optional[dict] = None,
+    ):
+        """The SQLite implementation ranks over the non-null values, so it needs their count."""
+        dependencies: dict = super()._get_evaluation_dependencies(
+            metric=metric,
+            configuration=configuration,
+            execution_engine=execution_engine,
+            runtime_configuration=runtime_configuration,
+        )
+
+        # Only SQLite reads this, so the other dialects are not charged an extra aggregate.
+        if (
+            isinstance(execution_engine, SqlAlchemyExecutionEngine)
+            and execution_engine.dialect_name == GXSqlDialect.SQLITE
+        ):
+            dependencies["column_values.nonnull.count"] = MetricConfiguration(
+                metric_name="column_values.nonnull.count",
+                metric_domain_kwargs=metric.metric_domain_kwargs,
+            )
+
+        return dependencies
+
 
 def _get_column_quantiles_sql_server(
     column, quantiles: Iterable, selectable, execution_engine: SqlAlchemyExecutionEngine
@@ -247,6 +290,7 @@ def _get_column_quantiles_mysql(
                 sa.dialects.mysql.DECIMAL(18, 15),
             ).label("p"),
         )
+        .where(column != None)  # noqa: E711 # FIXME CoP
         .order_by(sa.column("p").asc())
         .select_from(selectable)
         .cte("t")
@@ -278,7 +322,12 @@ def _get_column_quantiles_mysql(
 
     try:
         quantiles_results = execution_engine.execute_query(quantiles_query).fetchone()
-        return list(quantiles_results)  # type: ignore[arg-type] # FIXME CoP
+        # Filtering the nulls out of the CTE leaves it empty for a column with no non-null values,
+        # and the query then returns no row at all. Report one absent quantile per requested
+        # quantile, which is what the engines with a native "percentile_disc" return.
+        if quantiles_results is None:
+            return [None] * len(selects)
+        return list(quantiles_results)
     except sqlalchemy.ProgrammingError as pe:
         exception_message: str = "An SQL syntax Exception occurred."
         exception_traceback: str = traceback.format_exc()
@@ -330,18 +379,36 @@ def _get_column_quantiles_sqlite(
     quantiles: Iterable,
     selectable,
     execution_engine: SqlAlchemyExecutionEngine,
-    table_row_count,
+    nonnull_count: int,
 ) -> list:
     """
     The present implementation is somewhat inefficient, because it requires as many calls to
     "execution_engine.execute_query()" as the number of partitions in the "quantiles" parameter (albeit, typically,
     only a few).  However, this is the only mechanism available for SQLite at the present time (11/17/2021), because
     the analytical processing is not a very strongly represented capability of the SQLite database management system.
+
+    Ranks are taken over the non-null values only and follow "percentile_disc", which returns the
+    first value whose cumulative distribution reaches the quantile.
     """  # noqa: E501 # FIXME CoP
-    offsets: list[int] = [quantile * table_row_count - 1 for quantile in quantiles]
+    # The rank is "ceil(quantile * count)" evaluated on the quantile as written, not on its binary
+    # approximation. 0.56 is not representable in binary, so 0.56 * 25 is 14.000000000000002 and
+    # ceiling that selects rank 15. Rank 14 is the correct answer, because 14/25 is exactly 0.56
+    # and so already reaches the quantile.
+    #
+    # Note that this deliberately differs from the SQL engines that implement "percentile_disc" in
+    # double precision: PostgreSQL returns the 15th value here. The divergence is confined to
+    # quantiles whose product with the count is a whole number in decimal but not in binary.
+    ranks: list[int] = [
+        max(math.ceil(Fraction(str(quantile)) * nonnull_count), 1) for quantile in quantiles
+    ]
     quantile_queries: list[sqlalchemy.Select] = [
-        sa.select(column).order_by(column.asc()).offset(offset).limit(1).select_from(selectable)
-        for offset in offsets
+        sa.select(column)
+        .where(column != None)  # noqa: E711 # FIXME CoP
+        .order_by(column.asc())
+        .offset(rank - 1)
+        .limit(1)
+        .select_from(selectable)
+        for rank in ranks
     ]
 
     try:

tests/expectations/metrics/test_core.py

@@ -842,6 +842,34 @@ def test_quantiles_metric_sa(sa):
     results = engine.resolve_metrics(metrics_to_resolve=(table_row_count_metric,), metrics=metrics)
     metrics.update(results)
 
+    column_values_null_condition_metric = MetricConfiguration(
+        metric_name=f"column_values.null.{MetricPartialFunctionTypeSuffixes.CONDITION.value}",
+        metric_domain_kwargs={"column": "a"},
+        metric_value_kwargs=None,
+    )
+    column_values_null_condition_metric.metric_dependencies = {
+        "table.columns": table_columns_metric,
+    }
+    results = engine.resolve_metrics(
+        metrics_to_resolve=(column_values_null_condition_metric,), metrics=metrics
+    )
+    metrics.update(results)
+
+    column_values_nonnull_count_metric = MetricConfiguration(
+        metric_name=f"column_values.null.{SummarizationMetricNameSuffixes.UNEXPECTED_COUNT.value}",
+        metric_domain_kwargs={"column": "a"},
+        metric_value_kwargs=None,
+    )
+    column_values_nonnull_count_metric.metric_dependencies = {
+        "unexpected_condition": column_values_null_condition_metric,
+        "metric_partial_fn": partial_metric,
+        "table.columns": table_columns_metric,
+    }
+    results = engine.resolve_metrics(
+        metrics_to_resolve=(column_values_nonnull_count_metric,), metrics=metrics
+    )
+    metrics.update(results)
+
     desired_metric = MetricConfiguration(
         metric_name="column.quantile_values",
         metric_domain_kwargs={"column": "a"},
@@ -852,6 +880,7 @@ def test_quantiles_metric_sa(sa):
     desired_metric.metric_dependencies = {
         "table.columns": table_columns_metric,
         "table.row_count": table_row_count_metric,
+        "column_values.nonnull.count": column_values_nonnull_count_metric,
     }
     results = engine.resolve_metrics(metrics_to_resolve=(desired_metric,), metrics=metrics)
     metrics.update(results)

tests/integration/data_sources_and_expectations/expectations/test_expect_column_quantile_values_to_be_between.py

@@ -40,6 +40,19 @@
 
 DATA = pd.DataFrame({COL_NAME: [1, 2, 2, 3, 3, 3, 4]})
 
+# Same distribution as DATA, plus nulls. Excluding the nulls from the computation must yield the
+# same quantiles DATA produces. The dtype keeps the column an integer one on the SQL backends, so
+# that the nulls are the only thing that differs from DATA.
+DATA_WITH_NULLS = pd.DataFrame({COL_NAME: [1, 2, 2, 3, 3, 3, 4, None, None]}, dtype="object")
+
+# No duplicates, and quantiles chosen so that quantile * row_count is not a whole number, which is
+# where the selected rank is easiest to get wrong.
+DISTINCT_DATA = pd.DataFrame({COL_NAME: [10, 20, 30, 40]})
+
+# 0.56 is not representable in binary, so 0.56 * 25 is 14.000000000000002 and a rank computed in
+# floating point rounds up to 15. The correct rank is 14, because 14/25 is exactly 0.56.
+TWENTY_FIVE_ROWS = pd.DataFrame({COL_NAME: list(range(1, 26))})
+
 ALL_NULLS = pd.DataFrame({COL_NAME: [None, None, None]}, dtype="object")
 
 # An all-null column carries no type of its own, so each backend is told what to make it.
@@ -124,6 +137,88 @@ def test_all_null_column_reports_unmet_expectation(batch_for_datasource: Batch)
     }
 
 
+@parameterize_batch_for_data_sources(
+    data_source_configs=ALL_DATA_SOURCES_EXCEPT_BIGQUERY, data=DATA_WITH_NULLS
+)
+def test_nulls_are_excluded_from_quantiles(batch_for_datasource: Batch) -> None:
+    expectation = gxe.ExpectColumnQuantileValuesToBeBetween(
+        column=COL_NAME,
+        quantile_ranges=QuantileRange(
+            quantiles=[0, 0.333, 0.667, 1],
+            value_ranges=[[0, 1], [2, 3], [3, 4], [4, 5]],
+        ),
+    )
+    result = batch_for_datasource.validate(expectation, result_format=ResultFormat.COMPLETE)
+    assert result.success
+    assert result.to_json_dict()["result"] == {
+        "observed_value": {
+            "quantiles": [0.0, 0.333, 0.667, 1.0],
+            "values": [1, 2, 3, 4],
+        },
+        "details": {
+            "success_details": [True, True, True, True],
+        },
+    }
+
+
+# Pandas and SQLite only. The MySQL implementation resolves quantiles from percent_rank, which
+# picks the largest rank at or below the quantile rather than the first one to reach it, so it
+# reports [10, 20] for this data. That predates this change and is left alone here.
+@parameterize_batch_for_data_sources(
+    data_source_configs=[*JUST_PANDAS_DATA_SOURCES, SqliteDatasourceTestConfig()],
+    data=DISTINCT_DATA,
+)
+def test_quantiles_when_quantile_times_count_is_not_whole(
+    batch_for_datasource: Batch,
+) -> None:
+    expectation = gxe.ExpectColumnQuantileValuesToBeBetween(
+        column=COL_NAME,
+        quantile_ranges=QuantileRange(
+            quantiles=[0.3, 0.6],
+            value_ranges=[[20, 20], [30, 30]],
+        ),
+    )
+    result = batch_for_datasource.validate(expectation, result_format=ResultFormat.COMPLETE)
+    assert result.success
+    assert result.to_json_dict()["result"] == {
+        "observed_value": {
+            "quantiles": [0.3, 0.6],
+            "values": [20, 30],
+        },
+        "details": {
+            "success_details": [True, True],
+        },
+    }
+
+
+# Pandas and SQLite only, and deliberately so: do not extend this to the other SQL backends. They
+# implement "percentile_disc" in double precision, where 0.56 * 25 exceeds 14, so PostgreSQL and
+# its peers report 15 for this data. SQLite ranks on the quantile as written and reports 14.
+#
+# This case already passes on develop, where truncating the offset happens to land on 14. It is
+# here to pin the exact-decimal arithmetic, not to reproduce a defect.
+@parameterize_batch_for_data_sources(
+    data_source_configs=[*JUST_PANDAS_DATA_SOURCES, SqliteDatasourceTestConfig()],
+    data=TWENTY_FIVE_ROWS,
+)
+def test_quantile_times_count_inexact_in_binary_float(batch_for_datasource: Batch) -> None:
+    expectation = gxe.ExpectColumnQuantileValuesToBeBetween(
+        column=COL_NAME,
+        quantile_ranges=QuantileRange(quantiles=[0.56], value_ranges=[[14, 14]]),
+    )
+    result = batch_for_datasource.validate(expectation, result_format=ResultFormat.COMPLETE)
+    assert result.success
+    assert result.to_json_dict()["result"] == {
+        "observed_value": {
+            "quantiles": [0.56],
+            "values": [14],
+        },
+        "details": {
+            "success_details": [True],
+        },
+    }
+
+
 @parameterize_batch_for_data_sources(data_source_configs=JUST_PANDAS_DATA_SOURCES, data=DATA)
 def test_allows_unspecified_extremes(batch_for_datasource: Batch) -> None:
     expectation = gxe.ExpectColumnQuantileValuesToBeBetween(