diff --git a/great_expectations/expectations/metrics/column_aggregate_metrics/column_quantile_values.py b/great_expectations/expectations/metrics/column_aggregate_metrics/column_quantile_values.py index f5cf763a6f98..f4c904304acf 100644 --- a/great_expectations/expectations/metrics/column_aggregate_metrics/column_quantile_values.py +++ b/great_expectations/expectations/metrics/column_aggregate_metrics/column_quantile_values.py @@ -3,9 +3,11 @@ 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 @@ -13,8 +15,10 @@ 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: diff --git a/tests/expectations/metrics/test_core.py b/tests/expectations/metrics/test_core.py index 1455d1d0cb4d..107e8b9ce21d 100644 --- a/tests/expectations/metrics/test_core.py +++ b/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) diff --git a/tests/integration/data_sources_and_expectations/expectations/test_expect_column_quantile_values_to_be_between.py b/tests/integration/data_sources_and_expectations/expectations/test_expect_column_quantile_values_to_be_between.py index 3112c3ca692c..6a98cfc7f0b5 100644 --- a/tests/integration/data_sources_and_expectations/expectations/test_expect_column_quantile_values_to_be_between.py +++ b/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(