diff --git a/great_expectations/expectations/metrics/column_map_metrics/column_values_unique.py b/great_expectations/expectations/metrics/column_map_metrics/column_values_unique.py index e93d2f002284..0b895b4cee9f 100644 --- a/great_expectations/expectations/metrics/column_map_metrics/column_values_unique.py +++ b/great_expectations/expectations/metrics/column_map_metrics/column_values_unique.py @@ -1,6 +1,9 @@ from __future__ import annotations -from great_expectations.compatibility import pyspark +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union + +import great_expectations.exceptions as gx_exceptions +from great_expectations.compatibility import pyspark, sqlalchemy from great_expectations.compatibility.pyspark import functions as F from great_expectations.compatibility.sqlalchemy import ( Select, @@ -8,95 +11,318 @@ from great_expectations.compatibility.sqlalchemy import ( sqlalchemy as sa, ) -from great_expectations.core.metric_function_types import MetricPartialFunctionTypes +from great_expectations.compatibility.typing_extensions import override +from great_expectations.constants import MAX_RESULT_RECORDS +from great_expectations.core.metric_function_types import ( + MetricPartialFunctionTypes, + MetricPartialFunctionTypeSuffixes, +) from great_expectations.execution_engine import ( + ExecutionEngine, PandasExecutionEngine, SparkDFExecutionEngine, SqlAlchemyExecutionEngine, ) -from great_expectations.execution_engine.sqlalchemy_dialect import ( - GXSqlDialect, - quote_str, -) from great_expectations.expectations.metrics.map_metric_provider import ( ColumnMapMetricProvider, column_condition_partial, + column_function_partial, +) +from great_expectations.expectations.metrics.map_metric_provider.map_condition_auxilliary_methods import ( # noqa: E501 # long module path + _get_sqlalchemy_customized_unexpected_index_list, ) -from great_expectations.util import generate_temporary_table_name +from great_expectations.expectations.metrics.util import sqlalchemy_select_to_sql_string +from great_expectations.util import get_sqlalchemy_selectable +from great_expectations.validator.validation_graph import MetricConfiguration + +if TYPE_CHECKING: + from great_expectations.expectations.expectation_configuration import ( + ExpectationConfiguration, + ) + + +_DUP_KEY_COUNT_LABEL = "_num_rows" +_DUP_KEY_SUBQUERY_ALIAS = "column_values_count_per_value_subquery" + + +def _named_source_subquery(selectable, table_columns: List[str]): + """Return a named subquery that explicitly projects "table_columns" from + the source selectable. + + "SqlAlchemyBatchData" exposes the source table as a metadata-less + "sa.Table" shell (no reflected columns), so its ".c" accessor is empty. + Likewise, when a row_condition is present "get_domain_records" returns a + "SELECT * FROM ... WHERE ..." Select whose ".c" collection carries no named + columns. Wrapping either shape in an explicit projection gives us a + subquery whose ".c" collection is populated and can be used to + unambiguously reference source-side columns inside a join with the + dup-keys subquery. + """ + from_clause = selectable.subquery() if isinstance(selectable, Select) else selectable + return ( + sa.select(*[sa.column(c) for c in table_columns]) + .select_from(from_clause) + .subquery("column_values_unique_source") + ) + + +def _build_dup_keys_subquery( + execution_engine: SqlAlchemyExecutionEngine, + metric_domain_kwargs: Dict[str, Any], + column_name: str, +): + """Narrow GROUP BY/HAVING subquery: one row per duplicated value. + + Reads only the target column from the source table; partial-aggregation + friendly on distributed engines and avoids wide-row window sort. + """ + selectable = execution_engine.get_domain_records(domain_kwargs=metric_domain_kwargs) + selectable = get_sqlalchemy_selectable(selectable) # type: ignore[arg-type] # FIXME CoP + return ( + sa.select(sa.column(column_name)) + .select_from(selectable) # type: ignore[arg-type] # FIXME CoP + .where(sa.column(column_name).is_not(None)) + .group_by(sa.column(column_name)) + .having(sa.func.count() >= 2) # noqa: PLR2004 # 2 is the duplicate threshold + .subquery("column_values_unique_dup_keys") + ) + + +def _sqlalchemy_unique_unexpected_rows( + cls, + execution_engine: SqlAlchemyExecutionEngine, + metric_domain_kwargs: Dict[str, Any], + metric_value_kwargs: Dict[str, Any], + metrics: Dict[str, Any], + **kwargs, +) -> Sequence[Any]: + """Return full source rows for values that appear more than once. + + Source is scanned twice (cheap narrow hash-aggregate + hash join back), + but only when the caller requests "unexpected_rows" (typically COMPLETE + result_format). The dominant "unexpected_count" path stays single-scan. + """ + column_name: str = metric_domain_kwargs["column"] + table_columns: List[str] = metrics["table.columns"] + source_selectable = _named_source_subquery( + execution_engine.get_domain_records(domain_kwargs=metric_domain_kwargs), + table_columns, + ) + dup_keys = _build_dup_keys_subquery( + execution_engine=execution_engine, + metric_domain_kwargs=metric_domain_kwargs, + column_name=column_name, + ) + column_selector = [source_selectable.c[c] for c in table_columns] + query = sa.select(*column_selector).select_from( + source_selectable.join( + dup_keys, + source_selectable.c[column_name] == dup_keys.c[column_name], + ) + ) + result_format = metric_value_kwargs["result_format"] + if result_format["result_format"] != "COMPLETE": + limit = min(result_format["partial_unexpected_count"], MAX_RESULT_RECORDS) + query = query.limit(limit) + try: + return [ + row._asdict() + for row in execution_engine.execute_query(query).fetchmany(MAX_RESULT_RECORDS) + ] + except sqlalchemy.OperationalError as oe: + raise gx_exceptions.InvalidMetricAccessorDomainKwargsKeyError( + message=f"An SQL execution Exception occurred: {oe!s}." + ) + + +def _sqlalchemy_unique_unexpected_index_list( + cls, + execution_engine: SqlAlchemyExecutionEngine, + metric_domain_kwargs: Dict[str, Any], + metric_value_kwargs: Dict[str, Any], + metrics: Dict[str, Any], + **kwargs, +) -> Union[List[Dict[str, Any]], None]: + """Return specified index columns + target column for duplicate rows.""" + result_format = metric_value_kwargs["result_format"] + unexpected_index_column_names = result_format.get("unexpected_index_column_names") + if not unexpected_index_column_names: + return None + + column_name: str = metric_domain_kwargs["column"] + all_table_columns: List[str] = metrics.get("table.columns", []) + for idx_col in unexpected_index_column_names: + if idx_col not in all_table_columns: + raise gx_exceptions.InvalidMetricAccessorDomainKwargsKeyError( + message=( + f'Error: The unexpected_index_column: "{idx_col}" does not exist in ' + "SQL Table. Please check your configuration and try again." + ) + ) + + source_selectable = _named_source_subquery( + execution_engine.get_domain_records(domain_kwargs=metric_domain_kwargs), + all_table_columns, + ) + dup_keys = _build_dup_keys_subquery( + execution_engine=execution_engine, + metric_domain_kwargs=metric_domain_kwargs, + column_name=column_name, + ) + column_selector = [source_selectable.c[c] for c in unexpected_index_column_names] + column_selector.append(source_selectable.c[column_name]) + query = ( + sa.select(*column_selector) + .select_from( + source_selectable.join( + dup_keys, + source_selectable.c[column_name] == dup_keys.c[column_name], + ) + ) + .limit(result_format["partial_unexpected_count"]) + ) + exclude_unexpected_values: bool = result_format.get("exclude_unexpected_values", False) + try: + query_result: List[sqlalchemy.Row] = execution_engine.execute_query(query).fetchall() # type: ignore[assignment] # FIXME CoP + except sqlalchemy.OperationalError as oe: + raise gx_exceptions.InvalidMetricAccessorDomainKwargsKeyError( + message=f"An SQL execution Exception occurred: {oe!s}." + ) + + return _get_sqlalchemy_customized_unexpected_index_list( + exclude_unexpected_values=exclude_unexpected_values, + unexpected_index_column_names=unexpected_index_column_names, + query_result=query_result, + domain_column_name_list=[column_name], + ) + + +def _sqlalchemy_unique_unexpected_index_query( + cls, + execution_engine: SqlAlchemyExecutionEngine, + metric_domain_kwargs: Dict[str, Any], + metric_value_kwargs: Dict[str, Any], + metrics: Dict[str, Any], + **kwargs, +) -> Optional[str]: + """Return an executable SQL string selecting the duplicate rows. + + The default "_sqlalchemy_map_condition_query" renders the map condition + against the raw source table, but our condition references the narrow + count-per-value subquery, which is absent from that FROM clause. Build the + query string from the same join-back pattern used by the other + row-retrieval paths instead, so the string surfaced in validation results + and Data Docs runs against the source database as-is. + """ + result_format = metric_value_kwargs["result_format"] + if result_format.get("return_unexpected_index_query") is False: + return None + + column_name: str = metric_domain_kwargs["column"] + all_table_columns: List[str] = metrics.get("table.columns", []) + unexpected_index_column_names: List[str] = ( + result_format.get("unexpected_index_column_names") or [] + ) + for idx_col in unexpected_index_column_names: + if idx_col not in all_table_columns: + raise gx_exceptions.InvalidMetricAccessorDomainKwargsKeyError( + message=( + f'Error: The unexpected_index_column: "{idx_col}" does not exist in ' + "SQL Table. Please check your configuration and try again." + ) + ) + + source_selectable = _named_source_subquery( + execution_engine.get_domain_records(domain_kwargs=metric_domain_kwargs), + all_table_columns, + ) + dup_keys = _build_dup_keys_subquery( + execution_engine=execution_engine, + metric_domain_kwargs=metric_domain_kwargs, + column_name=column_name, + ) + column_selector = [source_selectable.c[c] for c in unexpected_index_column_names] + column_selector.append(source_selectable.c[column_name]) + query = sa.select(*column_selector).select_from( + source_selectable.join( + dup_keys, + source_selectable.c[column_name] == dup_keys.c[column_name], + ) + ) + return sqlalchemy_select_to_sql_string(engine=execution_engine, select_statement=query) class ColumnValuesUnique(ColumnMapMetricProvider): + """Detects duplicate values in a column. + + The "SqlAlchemyExecutionEngine" implementation materializes a *narrow* windowed + subquery that exposes only the target column and a "_num_rows" count per value. + Because the source table is scanned exactly once and the window operator carries + only one column through the sort/partition phase, this avoids both: + + * the "col NOT IN (dup_subquery)" double-scan pattern (original failure mode), + * the "SELECT *table_columns, count() OVER ... FROM source" wide-row window that + forced Redshift to materialize every column (including JSON/SUPER fields) + through the sort, occasionally tripping the WLM "low_timeout" rule on + column-store backends even after the double-scan was removed. + + Auxiliary metrics that need the full source row ("unexpected_rows") or specific + "unexpected_index_column_names" are served by a separate join-back path that + re-reads only the necessary columns from the source table, keeping the common + "BASIC" result_format (only "unexpected_count" requested) on the single-scan + fast path. + """ + + function_metric_name = "column_values.count_per_value" condition_metric_name = "column_values.unique" + # The narrow windowed subquery (below) carries only the target column. The + # default map-condition row-retrieval providers assume the selectable + # carries every table column (compound_columns.unique pattern), which + # would re-introduce the wide-row window on Redshift. Override the three + # SqlAlchemy row-retrieval hooks with narrow dup-keys subqueries joined + # back to source. + sqlalchemy_unexpected_rows_provider = staticmethod(_sqlalchemy_unique_unexpected_rows) + sqlalchemy_unexpected_index_list_provider = staticmethod( + _sqlalchemy_unique_unexpected_index_list + ) + sqlalchemy_unexpected_index_query_provider = staticmethod( + _sqlalchemy_unique_unexpected_index_query + ) + @column_condition_partial(engine=PandasExecutionEngine) def _pandas(cls, column, **kwargs): return ~column.duplicated(keep=False) - # NOTE: 20201119 - JPC - We cannot split per-dialect into window and non-window functions - # @column_condition_partial( - # engine=SqlAlchemyExecutionEngine, - # ) - # def _sqlalchemy(cls, column, _table, **kwargs): - # dup_query = ( - # sa.select(column) - # .select_from(_table) - # .group_by(column) - # .having(sa.func.count(column) > 1) - # ) - # - # return column.notin_(dup_query) + @column_function_partial(engine=SqlAlchemyExecutionEngine) + def _sqlalchemy_function(cls, column, _table, **kwargs): + # Narrow projection: only the target column and the window count per value. + # Auxiliary methods that consume this selectable (unexpected_count, + # unexpected_values, unexpected_value_counts) only ever read these two + # columns. Paths that need additional source columns are overridden via + # the sqlalchemy_*_provider class attributes to join back to source. + from_clause = _table.subquery() if isinstance(_table, Select) else _table + return ( + sa.select( + sa.column(column.name), + sa.func.count() + .over(partition_by=sa.column(column.name)) + .label(_DUP_KEY_COUNT_LABEL), + ) + .select_from(from_clause) + .alias(_DUP_KEY_SUBQUERY_ALIAS) + ) @column_condition_partial( engine=SqlAlchemyExecutionEngine, partial_fn_type=MetricPartialFunctionTypes.WINDOW_CONDITION_FN, ) - def _sqlalchemy_window(cls, column, _table, **kwargs): - # MySQL and SingleStore cannot reference a temp table more than once in the - # same query, and SingleStore disallows correlated subselects with GROUP BY. - # Create a temp table copy of the column to avoid both issues. - dialect = kwargs.get("_dialect") - sql_engine = kwargs.get("_sqlalchemy_engine") - execution_engine = kwargs.get("_execution_engine") - try: - dialect_name = dialect.dialect.name - except AttributeError: - try: - dialect_name = dialect.name - except AttributeError: - dialect_name = "" - if sql_engine and dialect and dialect_name in ("mysql", "singlestoredb"): - gx_dialect = GXSqlDialect(dialect_name) - quoted_col = quote_str(column.name, gx_dialect) - temp_table_name = generate_temporary_table_name() - if isinstance(_table, Select): - from_clause = _table.subquery().alias("tmp") - else: - from_clause = _table - source_query = sa.select(sa.column(column.name)).select_from(from_clause) - compiled = source_query.compile( - dialect=sql_engine.dialect, compile_kwargs={"literal_binds": True} - ) - temp_table_stmt = f"CREATE TEMPORARY TABLE {temp_table_name} AS {compiled}" - execution_engine.execute_query_in_transaction(sa.text(temp_table_stmt)) - # SingleStore cannot handle subselects with GROUP BY/HAVING inside - # expressions, so materialize duplicate values into a second temp table. - dup_table_name = generate_temporary_table_name() - dup_stmt = ( - f"CREATE TEMPORARY TABLE {dup_table_name} AS " - f"SELECT {quoted_col} FROM {temp_table_name} " - f"GROUP BY {quoted_col} HAVING count({quoted_col}) > 1" - ) - execution_engine.execute_query_in_transaction(sa.text(dup_stmt)) - dup_query = sa.select(column).select_from(sa.text(dup_table_name)) - else: - from_clause = _table.subquery() if isinstance(_table, Select) else _table - dup_query = ( - sa.select(column) - .select_from(from_clause) - .group_by(column) - .having(sa.func.count(column) > 1) - ) - return column.notin_(dup_query) + def _sqlalchemy_condition(cls, column, **kwargs): + metrics = kwargs.get("_metrics") + count_per_value_query, _, _ = metrics[ + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}" + ] + return count_per_value_query.c[_DUP_KEY_COUNT_LABEL] < 2 # noqa: PLR2004 # 2 is the duplicate threshold @column_condition_partial( engine=SparkDFExecutionEngine, @@ -104,3 +330,33 @@ def _sqlalchemy_window(cls, column, _table, **kwargs): ) def _spark(cls, column, **kwargs): return F.count(F.lit(1)).over(pyspark.Window.partitionBy(column)) <= 1 + + @classmethod + @override + def _get_evaluation_dependencies( + cls, + metric: MetricConfiguration, + configuration: Optional[ExpectationConfiguration] = None, + execution_engine: Optional[ExecutionEngine] = None, + runtime_configuration: Optional[dict] = None, + ): + dependencies: dict = super()._get_evaluation_dependencies( + metric=metric, + configuration=configuration, + execution_engine=execution_engine, + runtime_configuration=runtime_configuration, + ) + + if isinstance(execution_engine, SqlAlchemyExecutionEngine) and ( + metric.metric_name + == f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}" + ): + dependencies[ + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}" + ] = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs=metric.metric_domain_kwargs, + metric_value_kwargs=None, + ) + + return dependencies diff --git a/great_expectations/expectations/metrics/map_metric_provider/is_sqlalchemy_metric_selectable.py b/great_expectations/expectations/metrics/map_metric_provider/is_sqlalchemy_metric_selectable.py index 042ea24d4fc6..aa8bb0cb0cf7 100644 --- a/great_expectations/expectations/metrics/map_metric_provider/is_sqlalchemy_metric_selectable.py +++ b/great_expectations/expectations/metrics/map_metric_provider/is_sqlalchemy_metric_selectable.py @@ -14,6 +14,7 @@ SQLALCHEMY_SELECTABLE_METRICS: Set[str] = { "compound_columns.count", "compound_columns.unique", + "column_values.unique", } diff --git a/tests/execution_engine/test_sqlalchemy_execution_engine.py b/tests/execution_engine/test_sqlalchemy_execution_engine.py index deb3310ce369..b714e8bb0b73 100644 --- a/tests/execution_engine/test_sqlalchemy_execution_engine.py +++ b/tests/execution_engine/test_sqlalchemy_execution_engine.py @@ -1070,12 +1070,28 @@ def validate_tmp_tables(execution_engine): validate_tmp_tables(execution_engine=execution_engine) + count_per_value_metric = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + count_per_value_metric.metric_dependencies = { + "table.columns": table_columns_metric, + } + results = execution_engine.resolve_metrics( + metrics_to_resolve=(count_per_value_metric,), metrics=metrics + ) + metrics.update(results) + + validate_tmp_tables(execution_engine=execution_engine) + condition_metric = MetricConfiguration( metric_name=f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}", metric_domain_kwargs={"column": "a"}, metric_value_kwargs=None, ) condition_metric.metric_dependencies = { + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}": count_per_value_metric, # noqa: E501 # metric name exceeds line length "table.columns": table_columns_metric, } results = execution_engine.resolve_metrics( diff --git a/tests/expectations/metrics/test_core.py b/tests/expectations/metrics/test_core.py index 714b3b96bf5d..1455d1d0cb4d 100644 --- a/tests/expectations/metrics/test_core.py +++ b/tests/expectations/metrics/test_core.py @@ -2628,12 +2628,24 @@ def test_map_unique_column_exists_sa(sa): table_columns_metric, results = get_table_columns_metric(execution_engine=engine) metrics.update(results) + count_per_value_metric = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + count_per_value_metric.metric_dependencies = { + "table.columns": table_columns_metric, + } + results = engine.resolve_metrics(metrics_to_resolve=(count_per_value_metric,), metrics=metrics) + metrics.update(results) + condition_metric = MetricConfiguration( metric_name=f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}", metric_domain_kwargs={"column": "a"}, metric_value_kwargs=None, ) condition_metric.metric_dependencies = { + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}": count_per_value_metric, # noqa: E501 # metric name exceeds line length "table.columns": table_columns_metric, } results = engine.resolve_metrics(metrics_to_resolve=(condition_metric,), metrics=metrics) @@ -2741,12 +2753,22 @@ def test_map_unique_empty_query_sa(sa): metrics: dict table_columns_metric, metrics = get_table_columns_metric(execution_engine=engine) + count_per_value_metric = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + count_per_value_metric.metric_dependencies = {"table.columns": table_columns_metric} + results = engine.resolve_metrics(metrics_to_resolve=(count_per_value_metric,), metrics=metrics) + metrics.update(results) + condition_metric = MetricConfiguration( metric_name=f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}", metric_domain_kwargs={"column": "a"}, metric_value_kwargs=None, ) condition_metric.metric_dependencies = { + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}": count_per_value_metric, # noqa: E501 # metric name exceeds line length "table.columns": table_columns_metric, } results = engine.resolve_metrics(metrics_to_resolve=(condition_metric,), metrics=metrics) @@ -2768,6 +2790,167 @@ def test_map_unique_empty_query_sa(sa): assert results[desired_metric.id] == 0 +@pytest.mark.sqlite +def test_map_unique_unexpected_count_sql_shape_sa(sa): + """Regression test for Redshift WLM "low_timeout" fixes. + + The single most common path on SQLAlchemy is "unexpected_count" (BASIC + result_format). The generated SQL must: + + * scan the source table exactly once (no "col NOT IN (dup_subquery)" + semi-join pattern, which double-scans on column-store backends), + * carry only the target column through the window operator (the inner + windowed subquery must NOT project arbitrary table columns, because + that materializes every column — including JSON/SUPER fields on + Redshift — through the partition sort and was observed to trip the + WLM "low_timeout" rule even after the double-scan was removed). + """ + engine = build_sa_execution_engine( + pd.DataFrame({"a": [1, 2, 3, 3, None], "b": ["x", "y", "z", "z", "w"]}), + sa, + ) + + executed_sql: list[str] = [] + + @sa.event.listens_for(engine.engine, "before_cursor_execute") + def capture_sql(conn, cursor, statement, parameters, context, executemany): + executed_sql.append(statement) + + table_columns_metric: MetricConfiguration + metrics: dict + table_columns_metric, metrics = get_table_columns_metric(execution_engine=engine) + + count_per_value_metric = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + count_per_value_metric.metric_dependencies = {"table.columns": table_columns_metric} + results = engine.resolve_metrics(metrics_to_resolve=(count_per_value_metric,), metrics=metrics) + metrics.update(results) + + condition_metric = MetricConfiguration( + metric_name=f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + condition_metric.metric_dependencies = { + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}": count_per_value_metric, # noqa: E501 # metric name exceeds line length + "table.columns": table_columns_metric, + } + results = engine.resolve_metrics(metrics_to_resolve=(condition_metric,), metrics=metrics) + metrics.update(results) + + unexpected_count_metric = MetricConfiguration( + metric_name=f"column_values.unique.{SummarizationMetricNameSuffixes.UNEXPECTED_COUNT.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + unexpected_count_metric.metric_dependencies = { + "unexpected_condition": condition_metric, + "table.columns": table_columns_metric, + } + results = engine.resolve_metrics(metrics_to_resolve=(unexpected_count_metric,), metrics=metrics) + assert results[unexpected_count_metric.id] == 2 + + combined = " ".join(executed_sql) + combined_upper = combined.upper() + + # Single-pass: no semi-join double-scan. + assert "NOT IN" not in combined_upper, ( + "Duplicate detection must not rely on a NOT IN (dup_subquery) pattern " + "(double-scans source on column-store DBs like Redshift)." + ) + + # Window must still be used for the count path (single scan, no GROUP BY + # collapse, framework's SUM(CASE) wrapper relies on one row per source row). + assert "PARTITION BY" in combined_upper, ( + f"Expected windowed unique check for unexpected_count, got: {executed_sql}" + ) + + # Narrow projection: only target column "a" must appear inside the windowed + # subquery. Column "b" must NOT be projected through the window operator — + # that was the wide-row failure mode that kept tripping WLM "low_timeout". + assert ", b," not in combined and ", b " not in combined and ', "b"' not in combined, ( + "Inner windowed subquery must project only the target column; carrying " + "extra source columns through the window operator re-introduces the " + "wide-row sort that trips Redshift WLM `low_timeout`. " + f"Got SQL: {executed_sql}" + ) + + +@pytest.mark.sqlite +def test_map_unique_unexpected_rows_join_back_sa(sa): + """The "unexpected_rows" path must hydrate full source rows by joining a + narrow dup-keys aggregate back to the source — never by widening the + windowed subquery to carry every source column. + """ + engine = build_sa_execution_engine( + pd.DataFrame( + { + "a": [1, 2, 3, 3, None], + "b": ["x", "y", "z1", "z2", "w"], + "c": [10, 20, 30, 31, 40], + } + ), + sa, + ) + + executed_sql: list[str] = [] + + @sa.event.listens_for(engine.engine, "before_cursor_execute") + def capture_sql(conn, cursor, statement, parameters, context, executemany): + executed_sql.append(statement) + + table_columns_metric: MetricConfiguration + metrics: dict + table_columns_metric, metrics = get_table_columns_metric(execution_engine=engine) + + count_per_value_metric = MetricConfiguration( + metric_name=f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + count_per_value_metric.metric_dependencies = {"table.columns": table_columns_metric} + results = engine.resolve_metrics(metrics_to_resolve=(count_per_value_metric,), metrics=metrics) + metrics.update(results) + + condition_metric = MetricConfiguration( + metric_name=f"column_values.unique.{MetricPartialFunctionTypeSuffixes.CONDITION.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs=None, + ) + condition_metric.metric_dependencies = { + f"column_values.count_per_value.{MetricPartialFunctionTypeSuffixes.MAP.value}": count_per_value_metric, # noqa: E501 # metric name exceeds line length + "table.columns": table_columns_metric, + } + results = engine.resolve_metrics(metrics_to_resolve=(condition_metric,), metrics=metrics) + metrics.update(results) + + unexpected_rows_metric = MetricConfiguration( + metric_name=f"column_values.unique.{SummarizationMetricNameSuffixes.UNEXPECTED_ROWS.value}", + metric_domain_kwargs={"column": "a"}, + metric_value_kwargs={ + "result_format": {"result_format": "COMPLETE", "partial_unexpected_count": 20} + }, + ) + unexpected_rows_metric.metric_dependencies = { + "unexpected_condition": condition_metric, + "table.columns": table_columns_metric, + } + results = engine.resolve_metrics(metrics_to_resolve=(unexpected_rows_metric,), metrics=metrics) + + rows = results[unexpected_rows_metric.id] + duplicated_a = sorted(r["a"] for r in rows) + assert duplicated_a == [3, 3] + + combined_upper = " ".join(executed_sql).upper() + assert "GROUP BY" in combined_upper and "HAVING" in combined_upper, ( + "unexpected_rows must use a narrow GROUP BY/HAVING dup-keys subquery " + f"joined back to source; got: {executed_sql}" + ) + + @pytest.mark.spark def test_map_unique_column_exists_spark(spark_session): engine: SparkDFExecutionEngine = build_spark_engine( diff --git a/tests/expectations/metrics/test_metric_providers.py b/tests/expectations/metrics/test_metric_providers.py index 3b78a82dd240..49afa6d57a3e 100644 --- a/tests/expectations/metrics/test_metric_providers.py +++ b/tests/expectations/metrics/test_metric_providers.py @@ -16,6 +16,11 @@ SqlAlchemyExecutionEngine, ) from great_expectations.expectations import registry +from great_expectations.expectations.metrics.column_map_metrics.column_values_unique import ( + _sqlalchemy_unique_unexpected_index_list, + _sqlalchemy_unique_unexpected_index_query, + _sqlalchemy_unique_unexpected_rows, +) from great_expectations.expectations.metrics.map_metric_provider import ( ColumnMapMetricProvider, ColumnPairMapMetricProvider, @@ -27,11 +32,6 @@ from great_expectations.expectations.metrics.map_metric_provider.column_pair_condition_partial import ( # noqa: E501 # FIXME CoP column_pair_condition_partial, ) -from great_expectations.expectations.metrics.map_metric_provider.map_condition_auxilliary_methods import ( # noqa: E501 # FIXME CoP - _sqlalchemy_map_condition_index, - _sqlalchemy_map_condition_query, - _sqlalchemy_map_condition_rows, -) from great_expectations.expectations.metrics.map_metric_provider.multicolumn_condition_partial import ( # noqa: E501 # FIXME CoP multicolumn_condition_partial, ) @@ -270,16 +270,15 @@ def _spark(cls, column, **kwargs): assert index_query_fn is _custom_index_query -def test__column_values_unique__sqlalchemy_row_retrieval_providers_are_generic(mock_registry): +def test__column_values_unique__sqlalchemy_row_retrieval_providers_are_narrow(mock_registry): """Regression guard for `column_values.unique`'s SqlAlchemy row-retrieval providers. - `ColumnValuesUnique` does not currently override the `MapMetricProvider` row-retrieval - provider hooks (see `test__map_metric_provider__sqlalchemy_row_retrieval_provider_hooks`), so - it still resolves to the generic full-row builders today. Once it overrides - `sqlalchemy_unexpected_rows_provider` / `sqlalchemy_unexpected_index_list_provider` / - `sqlalchemy_unexpected_index_query_provider` with narrow, single-scan providers, this test - must be updated to assert identity against those custom providers instead -- turning it into - a guard against silently reverting to the generic (wide-row) providers. + `ColumnValuesUnique` overrides the `MapMetricProvider` row-retrieval provider hooks + (`sqlalchemy_unexpected_rows_provider` / `sqlalchemy_unexpected_index_list_provider` / + `sqlalchemy_unexpected_index_query_provider`) with narrow, single-scan providers that + join a dup-keys subquery back to source instead of dragging every table column + through the window sort. This test guards against silently reverting to the generic + (wide-row) providers. """ # FIXME CoP _, rows_fn = mock_registry.get_sqlalchemy_metric_provider( "column_values.unique.unexpected_rows" @@ -291,9 +290,9 @@ def test__column_values_unique__sqlalchemy_row_retrieval_providers_are_generic(m "column_values.unique.unexpected_index_query" ) - assert rows_fn is _sqlalchemy_map_condition_rows - assert index_list_fn is _sqlalchemy_map_condition_index - assert index_query_fn is _sqlalchemy_map_condition_query + assert rows_fn is _sqlalchemy_unique_unexpected_rows + assert index_list_fn is _sqlalchemy_unique_unexpected_index_list + assert index_query_fn is _sqlalchemy_unique_unexpected_index_query def test__column_pair_map_metric__registration(mock_registry): diff --git a/tests/integration/data_sources_and_expectations/expectations/test_expect_column_values_to_be_unique.py b/tests/integration/data_sources_and_expectations/expectations/test_expect_column_values_to_be_unique.py index 19fc7e2dadb8..6b21d266baf6 100644 --- a/tests/integration/data_sources_and_expectations/expectations/test_expect_column_values_to_be_unique.py +++ b/tests/integration/data_sources_and_expectations/expectations/test_expect_column_values_to_be_unique.py @@ -1,11 +1,19 @@ +import subprocess +import sys from unittest.mock import ANY import pandas as pd import pytest import great_expectations.expectations as gxe +from great_expectations.compatibility.sqlalchemy import sqlalchemy as sa +from great_expectations.core.expectation_validation_result import ( + ExpectationValidationResult, +) from great_expectations.core.result_format import ResultFormat +from great_expectations.datasource.fluent import SQLDatasource from great_expectations.datasource.fluent.interfaces import Batch +from great_expectations.expectations.row_conditions import Column from tests.integration.conftest import parameterize_batch_for_data_sources from tests.integration.data_sources_and_expectations.test_canonical_expectations import ( JUST_PANDAS_DATA_SOURCES, @@ -15,6 +23,7 @@ from tests.integration.test_utils.data_source_config import ( MySQLDatasourceTestConfig, PostgreSQLDatasourceTestConfig, + SqliteDatasourceTestConfig, ) UNIQUE_INTS = "unique_integers" @@ -208,3 +217,182 @@ def test_include_unexpected_rows_sql(batch_for_datasource: Batch) -> None: unexpected_rows_str = str(unexpected_rows_data) assert "3" in unexpected_rows_str assert "c" in unexpected_rows_str + + +# The tests below pin down result_format behaviors of ExpectColumnValuesToBeUnique on +# SQL backends that are easy to break when the underlying `column_values.unique` metric +# implementation changes: composition with row_condition, the executability of +# unexpected_index_query, and the shared output shape of exclude_unexpected_values. + +ROW_ID = "row_id" +FILTER_FLAG = "filter_flag" +DUPLICATE_VALUES = "duplicate_values" + +RESULT_FORMAT_GUARD_DATA = pd.DataFrame( + { + ROW_ID: [1, 2, 3, 4, 5], + FILTER_FLAG: [0, 1, 1, 1, 1], + DUPLICATE_VALUES: [200, 200, 300, 300, 400], + } +) + +RESULT_FORMAT_GUARD_DATA_SOURCES = [ + PostgreSQLDatasourceTestConfig(), + SqliteDatasourceTestConfig(), +] + + +def _assert_no_metric_exceptions(result: ExpectationValidationResult) -> None: + """Fail loudly if any metric raised instead of producing a value.""" + exception_info = result.exception_info or {} + if "raised_exception" in exception_info: + assert not exception_info["raised_exception"], exception_info + else: + for info in exception_info.values(): + assert not (info or {}).get("raised_exception"), info + + +@parameterize_batch_for_data_sources( + data_source_configs=RESULT_FORMAT_GUARD_DATA_SOURCES, data=RESULT_FORMAT_GUARD_DATA +) +def test_complete_with_row_condition_and_unexpected_index_column_names_sql( + batch_for_datasource: Batch, +) -> None: + """Row-retrieval result_format options must compose with row_condition. + + The value 200 is duplicated only when the filtered-out row is included, so correct + results prove both that the row_condition was applied and that the index list was + hydrated without error. + """ + expectation = gxe.ExpectColumnValuesToBeUnique( + column=DUPLICATE_VALUES, + row_condition=Column(FILTER_FLAG) > 0, + condition_parser="great_expectations", + ) + result = batch_for_datasource.validate( + expectation, + result_format={ + "result_format": "COMPLETE", + "unexpected_index_column_names": [ROW_ID], + }, + ) + + _assert_no_metric_exceptions(result) + assert not result.success + assert result.result["unexpected_count"] == 2 + unexpected_index_list = sorted( + result.result["unexpected_index_list"], key=lambda entry: entry[ROW_ID] + ) + assert unexpected_index_list == [ + {ROW_ID: 3, DUPLICATE_VALUES: 300}, + {ROW_ID: 4, DUPLICATE_VALUES: 300}, + ] + + +@parameterize_batch_for_data_sources( + data_source_configs=RESULT_FORMAT_GUARD_DATA_SOURCES, data=RESULT_FORMAT_GUARD_DATA +) +def test_include_unexpected_rows_with_row_condition_sql( + batch_for_datasource: Batch, +) -> None: + """include_unexpected_rows must compose with row_condition on SQL backends.""" + expectation = gxe.ExpectColumnValuesToBeUnique( + column=DUPLICATE_VALUES, + row_condition=Column(FILTER_FLAG) > 0, + condition_parser="great_expectations", + ) + result = batch_for_datasource.validate( + expectation, + result_format={"result_format": "SUMMARY", "include_unexpected_rows": True}, + ) + + _assert_no_metric_exceptions(result) + assert not result.success + unexpected_rows = sorted(result.result["unexpected_rows"], key=lambda row: row[ROW_ID]) + assert unexpected_rows == [ + {ROW_ID: 3, FILTER_FLAG: 1, DUPLICATE_VALUES: 300}, + {ROW_ID: 4, FILTER_FLAG: 1, DUPLICATE_VALUES: 300}, + ] + + +@parameterize_batch_for_data_sources( + data_source_configs=RESULT_FORMAT_GUARD_DATA_SOURCES, data=RESULT_FORMAT_GUARD_DATA +) +def test_unexpected_index_query_is_executable_sql( + batch_for_datasource: Batch, +) -> None: + """The unexpected_index_query surfaced to users must run against the source database + and return exactly the unexpected rows. + """ + expectation = gxe.ExpectColumnValuesToBeUnique(column=DUPLICATE_VALUES) + result = batch_for_datasource.validate( + expectation, + result_format={ + "result_format": "COMPLETE", + "unexpected_index_column_names": [ROW_ID], + }, + ) + + _assert_no_metric_exceptions(result) + assert not result.success + unexpected_index_query = result.result["unexpected_index_query"] + assert unexpected_index_query + + datasource = batch_for_datasource.datasource + assert isinstance(datasource, SQLDatasource) + with datasource.get_engine().connect() as connection: + query_results = connection.execute(sa.text(unexpected_index_query.rstrip(";"))).fetchall() + + # Query selects the index column(s) followed by the expectation's column. + assert sorted(tuple(row) for row in query_results) == [ + (1, 200), + (2, 200), + (3, 300), + (4, 300), + ] + + +@parameterize_batch_for_data_sources( + data_source_configs=RESULT_FORMAT_GUARD_DATA_SOURCES, data=RESULT_FORMAT_GUARD_DATA +) +def test_exclude_unexpected_values_returns_columnar_index_list_sql( + batch_for_datasource: Batch, +) -> None: + """With exclude_unexpected_values=True, all SQL map expectations return a single + columnar entry ({index_column: [values, ...]}) rather than one dict per row. + ExpectColumnValuesToBeUnique must match that shared shape. + """ + expectation = gxe.ExpectColumnValuesToBeUnique(column=DUPLICATE_VALUES) + result = batch_for_datasource.validate( + expectation, + result_format={ + "result_format": "COMPLETE", + "unexpected_index_column_names": [ROW_ID], + "exclude_unexpected_values": True, + }, + ) + + _assert_no_metric_exceptions(result) + assert not result.success + unexpected_index_list = result.result["unexpected_index_list"] + assert len(unexpected_index_list) == 1 + assert sorted(unexpected_index_list[0][ROW_ID]) == [1, 2, 3, 4] + assert DUPLICATE_VALUES not in unexpected_index_list[0] + + +@pytest.mark.timeout(30) # the subprocess pays full library import cost +@pytest.mark.unit +def test_import_does_not_emit_metric_reregistration_warnings() -> None: + """Importing great_expectations must not warn about metric providers being + overwritten. Such warnings indicate a metric (e.g. column_values.unique) is + registered more than once with different providers, and they surface on stderr for + every user of the library. + """ + completed = subprocess.run( + [sys.executable, "-c", "import great_expectations"], + capture_output=True, + text=True, + check=True, + ) + assert "overwriting metric_provider" not in completed.stderr + assert "is being registered with different metric_provider" not in completed.stderr