Repository: great-expectations/great_expectations
Pull request #11272 (2025-06-25 18:46:05)

[BUGFIX] B/gx 1174/generalize schema expectation


- [ ] Description of PR changes above includes a link to [an existing GitHub issue](https://github.com/great-expectations/great_expectations/issues)
- [ ] PR title is prefixed with one of: [BUGFIX], [FEATURE], [DOCS], [MAINTENANCE], [CONTRIB], [MINORBUMP]
- [ ] Code is linted - run `invoke lint` (uses `ruff format` + `ruff check`)
- [ ] Appropriate tests and docs have been updated

For more information about contributing, visit our [community resources](https://docs.greatexpectations.io/docs/core/introduction/community_resources#contribute-code-or-documentation).

After you submit your PR, keep the page open and **monitor the statuses of the various checks made by our continuous integration process at the bottom of the page. Please fix any issues that come up** and [reach out on Slack](https://greatexpectations.io/slack) if you need help. Thanks for contributing!



Patch:
diff --git a/great_expectations/expectations/core/expect_table_columns_to_match_set.py b/great_expectations/expectations/core/expect_table_columns_to_match_set.py
index 357ea19f6..e90311c9c 100644
--- a/great_expectations/expectations/core/expect_table_columns_to_match_set.py
+++ b/great_expectations/expectations/core/expect_table_columns_to_match_set.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Type, Union
+from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Set, Type, Union
 
 from great_expectations.compatibility import pydantic
 from great_expectations.compatibility.typing_extensions import override
@@ -8,7 +8,6 @@ from great_expectations.core.suite_parameters import (
     SuiteParameterDict,  # noqa: TC001 # FIXME CoP
 )
 from great_expectations.exceptions.exceptions import InvalidSetTypeError
-from great_expectations.execution_engine.sqlalchemy_dialect import GXSqlDialect
 from great_expectations.expectations.expectation import (
     BatchExpectation,
     render_suite_parameter_string,
@@ -42,7 +41,11 @@ if TYPE_CHECKING:
 
 
 EXPECTATION_SHORT_DESCRIPTION = "Expect the columns in a table to match an unordered set."
-COLUMN_SET_DESCRIPTION = "The column names, in any order."
+COLUMN_SET_DESCRIPTION = (
+    "The column names, in any order. In SQL datasources, if the column names are "
+    "double quoted, for example '\"column_name\"', a case sensitive match is "
+    "done. Otherwise a case insensitive match is done."
+)
 EXACT_MATCH_DESCRIPTION = (
     "If True, the list of columns must exactly match the observed columns. "
     "If False, observed columns must include column_set but additional columns will pass."
@@ -423,111 +426,156 @@ class ExpectTableColumnsToMatchSet(BatchExpectation):
         runtime_configuration: Optional[dict] = None,
         execution_engine: Optional[ExecutionEngine] = None,
     ):
-        expected_column_set = _make_column_set_with_execution_engine_type(
-            self._get_success_kwargs().get("column_set"), execution_engine
+        from great_expectations.execution_engine import SqlAlchemyExecutionEngine
+
+        if isinstance(execution_engine, SqlAlchemyExecutionEngine):
+            return self._validate_sqlalchemy(metrics)
+
+        # Retrieve expected and observed column names
+        expected_column_list = self._get_success_kwargs().get("column_set")
+        expected_column_set = (
+            set(expected_column_list) if expected_column_list is not None else set()
         )
         actual_column_list = metrics.get("table.columns")
         actual_column_set = set(actual_column_list)
-        exact_match = self._get_success_kwargs().get("exact_match")
 
-        if (
-            (expected_column_set is None) and (exact_match is not True)
-        ) or actual_column_set == expected_column_set:
-            return {"success": True, "result": {"observed_value": actual_column_list}}
-        else:
-            # Convert to lists and sort to lock order for testing and output rendering
-            # unexpected_list contains items from the dataset columns that are not in expected_column_set  # noqa: E501 # FIXME CoP
-            unexpected_list = sorted(list(actual_column_set - expected_column_set))
-            # missing_list contains items from expected_column_set that are not in the dataset columns  # noqa: E501 # FIXME CoP
-            missing_list = sorted(list(expected_column_set - actual_column_set))
-            # observed_value contains items that are in the dataset columns
-            observed_value = sorted(actual_column_list)
-
-            mismatched = {}
-            if len(unexpected_list) > 0:
-                mismatched["unexpected"] = unexpected_list
-            if len(missing_list) > 0:
-                mismatched["missing"] = missing_list
+        unmatched_actual_column_set = actual_column_set - expected_column_set
+        unmatched_expected_column_set = expected_column_set - actual_column_set
+        return _validate_result(
+            actual_column_set,
+            expected_column_set,
+            unmatched_actual_column_set,
+            unmatched_expected_column_set,
+            self._get_success_kwargs().get("exact_match"),
+        )
 
-            result = {
-                "observed_value": observed_value,
-                "details": {"mismatched": mismatched},
-            }
+    def _validate_sqlalchemy(self, metrics: Dict):
+        # We want to match the expected columns with the actual columns. We first break up the
+        # expected columns into 2 sets, the quoted columns which must match exactly and the unquoted
+        # columns, which we case insensitive match.
+        expected_column_set = set(self._get_success_kwargs().get("column_set"))
+        quoted_expected_column_set = set()
+        unquoted_expected_column_set = set()
+        for col in expected_column_set:
+            if col.startswith('"') and col.endswith('"'):
+                quoted_expected_column_set.add(col[1:-1])
+            else:
+                unquoted_expected_column_set.add(col)
+
+        # The actual columns from the db will be unquoted and may be strs or CaseInsensitiveStrings.
+        # We normalize the actual_column_list to CaseInsensitiveStrings so we can use set operations
+        # going forward.
+        actual_column_list = metrics.get("table.columns")
+        actual_column_set = _make_case_insensitive_set(actual_column_list)
+
+        # We make copies of the expected and actual column sets and remove items from them as we
+        # find matches between the 2 sets.
+        unmatched_expected_column_set = expected_column_set.copy()
+        unmatched_actual_column_set = actual_column_set.copy()
+
+        # We first match quoted strings. The expected set is a set of strs while the actual set
+        # is a set of CaseInsensitiveStrings so we can't use set operations.
+        for col in actual_column_set:
+            if str(col) in quoted_expected_column_set:
+                unmatched_expected_column_set.remove(f'"{col!s}"')
+                unmatched_actual_column_set.remove(col)
+
+        # We normalize the unmatched_expected_column_set to CaseInsensitiveStrings
+        unmatched_expected_column_set = _make_case_insensitive_set(unmatched_expected_column_set)
+
+        # We now do the unquoted match
+        unquoted_expected_column_set = _make_case_insensitive_set(unquoted_expected_column_set)
+        unquoted_matches = unquoted_expected_column_set.intersection(unmatched_actual_column_set)
+
+        # We subtract the unquoted matches from the current unmatched sets to finalize them
+        unmatched_actual_column_set = unmatched_actual_column_set - unquoted_matches
+        unmatched_expected_column_set = unmatched_expected_column_set - unquoted_matches
+
+        return _validate_result(
+            actual_column_set,
+            expected_column_set,
+            unmatched_actual_column_set,
+            unmatched_expected_column_set,
+            self._get_success_kwargs().get("exact_match"),
+        )
 
-            return_success = {
-                "success": True,
-                "result": result,
-            }
-            return_failed = {
-                "success": False,
-                "result": result,
-            }
 
-            if exact_match:
-                return return_failed
-            else:  # noqa: PLR5501 # FIXME CoP
-                # Failed if there are items in the missing list (but OK to have unexpected_list)
-                if len(missing_list) > 0:
-                    return return_failed
-                # Passed if there are no items in the missing list
-                else:
-                    return return_success
-
-
-def _make_column_set_with_execution_engine_type(
-    column_set: Optional[set[str]],
-    execution_engine: Optional[ExecutionEngine],
-) -> set[str]:
+def _make_case_insensitive_set(
+    str_set: Optional[set[str | CaseInsensitiveString]],
+) -> set[CaseInsensitiveString]:
     """
-    Transforms column names in column_set to the appropriate type for the execution_engine.
+    Transforms a set of strs to CaseInsensitiveStrings.
 
     Args:
-        column_set: A set of column names.
-        execution_engine: An execution engine.
+        str_set: A set of strs.
 
     Returns:
-        A set of column names whose type matches the metric table.columns. This type varies
-        based on the execution engine.
+        A set of CaseInsensitiveString.
     """
-    if column_set is None:
-        return set()
-
-    dialect = (
-        execution_engine.dialect.name if execution_engine and execution_engine.dialect else None
+    from great_expectations.expectations.metrics.util import (
+        CaseInsensitiveString,
     )
-    if dialect and dialect in [
-        GXSqlDialect.DATABRICKS,
-        GXSqlDialect.POSTGRESQL,
-        GXSqlDialect.SNOWFLAKE,
-    ]:
-        # For these dialects, column_set we want to use
-        # The metric column_set will return a set of strs but table.columns will return a list of
-        # CaseInsensitiveStrings. CaseInsensitiveStrings and strs can be equal but have different
-        # hashes which breaks set operations. Since we want to do set operations in a case
-        # insensitive manner, we make the expect_column_set case insensitive.
-        return _make_case_insensitive_set(column_set)
-    else:
-        return set(column_set)
-
 
-def _make_case_insensitive_set(strs: set[str]) -> set[CaseInsensitiveString]:
-    """Creates a set of CaseInsensitiveStrings from a set of strs.
-
-    Args:
-        strs: A set of strs (not CaseInsensitiveStrings).
-
-    Returns:
-        A set of CaseInsensitiveStrings.
-    """
-    # Making a CaseInsentitiveSet data structure would be a more general solution and
-    # if we need to do this type transformation more than once we should make one in
-    # great_expectations/expectations/metrics/util.py.
-    from great_expectations.expectations.metrics.util import CaseInsensitiveString
+    if str_set is None:
+        return set()
 
     case_insensitive_strs = set()
-    for s in strs:
-        if isinstance(s, str) and not isinstance(s, CaseInsensitiveString):
+    for s in str_set:
+        if isinstance(s, CaseInsensitiveString):
+            case_insensitive_strs.add(s)
+        elif isinstance(s, str):
             case_insensitive_strs.add(CaseInsensitiveString(s))
         else:
-            raise InvalidSetTypeError("str", str(type(s)))
+            raise InvalidSetTypeError(
+                expected_type="str or CaseInsensitiveString", actual_type=str(type(s))
+            )
     return case_insensitive_strs
+
+
+def _validate_result(
+    actual_column_set: Set[Union[str, CaseInsensitiveString]],
+    expected_column_set: Set[str],
+    unmatched_actual_column_set: Set[Union[str, CaseInsensitiveString]],
+    unmatched_expected_column_set: Set[Union[str, CaseInsensitiveString]],
+    exact_match: bool,
+) -> Dict[str, Any]:
+    empty_set = set()
+    observed_value = sorted(list(actual_column_set))
+
+    if ((expected_column_set is None) and (exact_match is not True)) or (
+        unmatched_expected_column_set == empty_set and unmatched_actual_column_set == empty_set
+    ):
+        return {"success": True, "result": {"observed_value": observed_value}}
+    else:
+        unexpected_list = sorted(list(unmatched_actual_column_set))
+        missing_list = sorted(list(unmatched_expected_column_set))
+
+        mismatched = {}
+        if len(unexpected_list) > 0:
+            mismatched["unexpected"] = unexpected_list
+        if len(missing_list) > 0:
+            mismatched["missing"] = missing_list
+
+        result = {
+            "observed_value": observed_value,
+            "details": {"mismatched": mismatched},
+        }
+
+        return_success = {
+            "success": True,
+            "result": result,
+        }
+        return_failed = {
+            "success": False,
+            "result": result,
+        }
+
+        if exact_match:
+            return return_failed
+        else:  # noqa: PLR5501 # FIXME CoP
+            # Failed if there are items in the missing list (but OK to have unexpected_list)
+            if len(missing_list) > 0:
+                return return_failed
+            # Passed if there are no items in the missing list
+            else:
+                return return_success
diff --git a/great_expectations/expectations/core/schemas/ExpectTableColumnsToMatchSet.json b/great_expectations/expectations/core/schemas/ExpectTableColumnsToMatchSet.json
index 045552030..3cd13e31c 100644
--- a/great_expectations/expectations/core/schemas/ExpectTableColumnsToMatchSet.json
+++ b/great_expectations/expectations/core/schemas/ExpectTableColumnsToMatchSet.json
@@ -1,6 +1,6 @@
 {
     "title": "Expect table columns to match set",
-    "description": "Expect the columns in a table to match an unordered set.\n\nExpectTableColumnsToMatchSet is a     Batch Expectation.\n\nBatchExpectations are one of the most common types of Expectation.\nThey are evaluated for an entire Batch, and answer a semantic question about the Batch itself.\n\nArgs:\n    column_set (list of str): The column names, in any order.\n    exact_match (boolean):             If True, the list of columns must exactly match the observed columns. If False, observed columns must include column_set but additional columns will pass. Default True.\n\nOther Parameters:\n    result_format (str or None):             Which output mode to use: BOOLEAN_ONLY, BASIC, COMPLETE, or SUMMARY.             For more detail, see [result_format](https://docs.greatexpectations.io/docs/reference/expectations/result_format).\n    catch_exceptions (boolean or None):             If True, then catch exceptions and include them as part of the result object.             For more detail, see [catch_exceptions](https://docs.greatexpectations.io/docs/reference/expectations/standard_arguments/#catch_exceptions).\n    meta (dict or None):             A JSON-serializable dictionary (nesting allowed) that will be included in the output without             modification. For more detail, see [meta](https://docs.greatexpectations.io/docs/reference/expectations/standard_arguments/#meta).\n\nReturns:\n    An [ExpectationSuiteValidationResult](https://docs.greatexpectations.io/docs/terms/validation_result)\n\n    Exact fields vary depending on the values passed to result_format, catch_exceptions, and meta.\n\nSupported Data Sources:\n    [Pandas](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Spark](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [SQLite](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [PostgreSQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [MySQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [MSSQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [BigQuery](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Snowflake](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Databricks (SQL)](https://docs.greatexpectations.io/docs/application_integration_support/)\n\nData Quality Issues:\n    Schema\n\nExample Data:\n            test    test2\n        0   1.00    2\n        1   2.30    5\n        2   4.33    0\n\nCode Examples:\n    Passing Case:\n        Input:\n            ExpectTableColumnsToMatchSet(\n                column_set=[\"test\"],\n                exact_match=False\n        )\n\n        Output:\n            {\n              \"exception_info\": {\n                \"raised_exception\": false,\n                \"exception_traceback\": null,\n                \"exception_message\": null\n              },\n              \"result\": {\n                \"observed_value\": [\n                  \"test\",\n                  \"test2\"\n                ],\n                \"details\": {\n                  \"mismatched\": {\n                    \"unexpected\": [\n                      \"test2\"\n                    ]\n                  }\n                }\n              },\n              \"meta\": {},\n              \"success\": true\n            }\n\n    Failing Case:\n        Input:\n            ExpectTableColumnsToMatchSet(\n                column_set=[\"test2\", \"test3\"],\n                exact_match=True\n        )\n\n        Output:\n            {\n              \"exception_info\": {\n                \"raised_exception\": false,\n                \"exception_traceback\": null,\n                \"exception_message\": null\n              },\n              \"result\": {\n                \"observed_value\": [\n                  \"test\",\n                  \"test2\"\n                ],\n                \"details\": {\n                  \"mismatched\": {\n                    \"unexpected\": [\n                      \"test\"\n                    ],\n                    \"missing\": [\n                      \"test3\"\n                    ]\n                  }\n                }\n              },\n              \"meta\": {},\n              \"success\": false\n            }",
+    "description": "Expect the columns in a table to match an unordered set.\n\nExpectTableColumnsToMatchSet is a     Batch Expectation.\n\nBatchExpectations are one of the most common types of Expectation.\nThey are evaluated for an entire Batch, and answer a semantic question about the Batch itself.\n\nArgs:\n    column_set (list of str): The column names, in any order. In SQL datasources, if the column names are double quoted, for example '\"column_name\"', a case sensitive match is done. Otherwise a case insensitive match is done.\n    exact_match (boolean):             If True, the list of columns must exactly match the observed columns. If False, observed columns must include column_set but additional columns will pass. Default True.\n\nOther Parameters:\n    result_format (str or None):             Which output mode to use: BOOLEAN_ONLY, BASIC, COMPLETE, or SUMMARY.             For more detail, see [result_format](https://docs.greatexpectations.io/docs/reference/expectations/result_format).\n    catch_exceptions (boolean or None):             If True, then catch exceptions and include them as part of the result object.             For more detail, see [catch_exceptions](https://docs.greatexpectations.io/docs/reference/expectations/standard_arguments/#catch_exceptions).\n    meta (dict or None):             A JSON-serializable dictionary (nesting allowed) that will be included in the output without             modification. For more detail, see [meta](https://docs.greatexpectations.io/docs/reference/expectations/standard_arguments/#meta).\n\nReturns:\n    An [ExpectationSuiteValidationResult](https://docs.greatexpectations.io/docs/terms/validation_result)\n\n    Exact fields vary depending on the values passed to result_format, catch_exceptions, and meta.\n\nSupported Data Sources:\n    [Pandas](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Spark](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [SQLite](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [PostgreSQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [MySQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [MSSQL](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [BigQuery](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Snowflake](https://docs.greatexpectations.io/docs/application_integration_support/)\n    [Databricks (SQL)](https://docs.greatexpectations.io/docs/application_integration_support/)\n\nData Quality Issues:\n    Schema\n\nExample Data:\n            test    test2\n        0   1.00    2\n        1   2.30    5\n        2   4.33    0\n\nCode Examples:\n    Passing Case:\n        Input:\n            ExpectTableColumnsToMatchSet(\n                column_set=[\"test\"],\n                exact_match=False\n        )\n\n        Output:\n            {\n              \"exception_info\": {\n                \"raised_exception\": false,\n                \"exception_traceback\": null,\n                \"exception_message\": null\n              },\n              \"result\": {\n                \"observed_value\": [\n                  \"test\",\n                  \"test2\"\n                ],\n                \"details\": {\n                  \"mismatched\": {\n                    \"unexpected\": [\n                      \"test2\"\n                    ]\n                  }\n                }\n              },\n              \"meta\": {},\n              \"success\": true\n            }\n\n    Failing Case:\n        Input:\n            ExpectTableColumnsToMatchSet(\n                column_set=[\"test2\", \"test3\"],\n                exact_match=True\n        )\n\n        Output:\n            {\n              \"exception_info\": {\n                \"raised_exception\": false,\n                \"exception_traceback\": null,\n                \"exception_message\": null\n              },\n              \"result\": {\n                \"observed_value\": [\n                  \"test\",\n                  \"test2\"\n                ],\n                \"details\": {\n                  \"mismatched\": {\n                    \"unexpected\": [\n                      \"test\"\n                    ],\n                    \"missing\": [\n                      \"test3\"\n                    ]\n                  }\n                }\n              },\n              \"meta\": {},\n              \"success\": false\n            }",
     "type": "object",
     "properties": {
         "id": {
@@ -68,7 +68,7 @@
         },
         "column_set": {
             "title": "Column Set",
-            "description": "The column names, in any order.",
+            "description": "The column names, in any order. In SQL datasources, if the column names are double quoted, for example '\"column_name\"', a case sensitive match is done. Otherwise a case insensitive match is done.",
             "anyOf": [
                 {
                     "type": "array",


Test patch:
diff --git a/tests/integration/data_sources_and_expectations/expectations/test_expect_table_columns_to_match_set.py b/tests/integration/data_sources_and_expectations/expectations/test_expect_table_columns_to_match_set.py
index f222aacc9..cadcc365d 100644
--- a/tests/integration/data_sources_and_expectations/expectations/test_expect_table_columns_to_match_set.py
+++ b/tests/integration/data_sources_and_expectations/expectations/test_expect_table_columns_to_match_set.py
@@ -1,3 +1,5 @@
+from typing import Sequence
+
 import pandas as pd
 import pytest
 
@@ -8,12 +10,17 @@ from tests.integration.conftest import parameterize_batch_for_data_sources
 from tests.integration.data_sources_and_expectations.test_canonical_expectations import (
     ALL_DATA_SOURCES,
     JUST_PANDAS_DATA_SOURCES,
-)
-from tests.integration.test_utils.data_source_config.databricks import (
+    SQL_DATA_SOURCES,
+    BigQueryDatasourceTestConfig,
     DatabricksDatasourceTestConfig,
+    DataSourceTestConfig,
+    MSSQLDatasourceTestConfig,
+    MySQLDatasourceTestConfig,
+    PostgreSQLDatasourceTestConfig,
+    RedshiftDatasourceTestConfig,
+    SnowflakeDatasourceTestConfig,
+    SqliteDatasourceTestConfig,
 )
-from tests.integration.test_utils.data_source_config.postgres import PostgreSQLDatasourceTestConfig
-from tests.integration.test_utils.data_source_config.snowflake import SnowflakeDatasourceTestConfig
 
 COL_A = "col_a"
 COL_B = "col_b"
@@ -127,14 +134,133 @@ CASE_INSENSITIVE_DATA = pd.DataFrame(
 
 
 @parameterize_batch_for_data_sources(
-    data_source_configs=[
-        PostgreSQLDatasourceTestConfig(),
-        DatabricksDatasourceTestConfig(),
-        SnowflakeDatasourceTestConfig(),
-    ],
+    data_source_configs=SQL_DATA_SOURCES,
     data=CASE_INSENSITIVE_DATA,
 )
 def test_case_insensitive_success(batch_for_datasource: Batch) -> None:
     expectation = gxe.ExpectTableColumnsToMatchSet(column_set=["COLUMN_A", "column_b", "COLumN_c"])
     result = batch_for_datasource.validate(expectation)
     assert result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_case_insensitive_failure(batch_for_datasource: Batch) -> None:
+    expectation = gxe.ExpectTableColumnsToMatchSet(column_set=["COLUMN_Az", "column_b", "COLumN_c"])
+    result = batch_for_datasource.validate(expectation)
+    assert not result.success
+
+
+# For most of our tests we use all sql datasources. However, for some of them we exclude
+# snowflake and redshift for the following reasons:
+##### Snowflake #####
+# In test setup, sqlalchemy's CREATE TABLE will not quote lowercase column names
+# but will quote uppercase. Since snowflake stores case insensitive strings
+# as uppercase, while all the other databases we are testing stores them as lowercase
+# this creates different case insensitive columns. We break out the snowflake tests
+# into their own tests.
+#
+#### Redshift ####
+# Redshift by default is case insensitive so trying to match case via quoted identifiers
+# fails. There is a global setting you can set on a redshift cluster,
+# enable_case_sensitive_identifier, one can apply to make it case sensitive. Our code
+# looks like it would handle this successfully but I haven't verified since I need to
+# change our redshift CI cluster. There is tracked in GX-1197.
+
+SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT: Sequence[DataSourceTestConfig] = [
+    BigQueryDatasourceTestConfig(),
+    DatabricksDatasourceTestConfig(),
+    MSSQLDatasourceTestConfig(),
+    MySQLDatasourceTestConfig(),
+    PostgreSQLDatasourceTestConfig(),
+    SqliteDatasourceTestConfig(),
+]
+
+
+@pytest.mark.unit
+def test_sql_data_sources_without_snowflake_redshift() -> None:
+    # Verify SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT is what is says it is
+    assert len(SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT) + 2 == len(SQL_DATA_SOURCES)
+    assert SnowflakeDatasourceTestConfig() not in SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT
+    assert RedshiftDatasourceTestConfig() not in SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT
+    for datasource in SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT:
+        assert datasource in SQL_DATA_SOURCES
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_quoted_success(batch_for_datasource: Batch) -> None:
+    # When we create the table in the testing fr
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=['"column_a"', '"COLUMN_B"', '"CoLuMn_C"']
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=[SnowflakeDatasourceTestConfig()],
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_quoted_success_snowflake(batch_for_datasource: Batch) -> None:
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=['"column_a"', '"CoLuMn_C"'],
+        exact_match=False,
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_quoted_failure(batch_for_datasource: Batch) -> None:
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=['"Column_a"', '"COLUMN_B"', '"CoLuMn_C"']
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert not result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES_WITHOUT_SNOWFLAKE_REDSHIFT,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_unquoted_and_quoted_success(batch_for_datasource: Batch) -> None:
+    # Column A and B are case insensitve
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=["Column_a", "COLUMN_B", '"CoLuMn_C"']
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_unquoted_and_quoted_with_unquoted_failure(batch_for_datasource: Batch) -> None:
+    # Column A and B are case insensitve
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=["Column_az", "COLUMN_B", '"CoLuMn_C"']
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert not result.success
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=SQL_DATA_SOURCES,
+    data=CASE_INSENSITIVE_DATA,
+)
+def test_unquoted_and_quoted_with_quoted_failure(batch_for_datasource: Batch) -> None:
+    # Column C is case sensitve
+    expectation = gxe.ExpectTableColumnsToMatchSet(
+        column_set=["column_a", "COLUMN_B", '"Column_c"']
+    )
+    result = batch_for_datasource.validate(expectation)
+    assert not result.success