[MAINTENANCE] Add SQL result_format regression coverage for expect_column_values_to_be_unique

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