[BUGFIX] Narrow single-pass column_values.unique on SQLAlchemy (Redshift WLM)

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",
 }
 
 

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(

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(

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):