Repository: great-expectations/great_expectations
Pull request: [WIP] [FEATURE] add backend args to run_diagnostics
Fixes #3224 

There are many features that are supported in most SQL dialects but not sqlite, it would be useful to have tests for other dialects set to run in Expectation.run_diagnostics. 

I welcome suggestions for a cleaner implementation, but because the backends list is currently constructed before iterating over the examples, it is not straightforward to include this in Expectation.examples['tests']

Changes proposed in this pull request:
- add "execution_engine_args" to run_diagnostics that's passed through to build_test_backends_list
- add an example using the contrib column skew expectation

### Definition of Done
Please delete options that are not relevant.

- [x] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/en/latest/contributing/style_guide.html?highlight=style%20guide)
- [x] I have performed a [self-review](https://docs.greatexpectations.io/en/latest/contributing/contribution_checklist.html?highlight=checklist) of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] I have made sure that existing tests are passing.
- [x] I have run any local integration tests and made sure that nothing is broken.


Thank you for submitting!


Problem statement:
Allow run_diagnostics to specify a different SQLAlchemy backend
**Is your feature request related to a problem? Please describe.**
Many computations aren't supported in sqlite (for example, I'm not sure the core standard deviation expectation works). It would be nice to set a different backend for the unit tests for these expectations, or at least easily add MySQL or Postgres to them. 

**Describe the solution you'd like**
Allow expectations to define test_backends_options that are passed to great_expectations/self_check/util.py::build_test_backends_list 

**Describe alternatives you've considered**
It's possible to live with things as-is but the supported feature set for sqlite is very different from other SQL dialects.

Hints from issue discussion:
Hi @edjoesu - thanks for raising this issue! This sounds like an excellent idea, though not something we'd be able to prioritize building internally right now. Is this something you'd have an interest in contributing? I'm happy to offer any guidance/support/review you'd need. Just let me know how I can help.
@talagluck -- thanks! Yeah, why don't I put together a sample PR and you can tell me if there's a cleaner way to do it. 
Thanks, @edjoesu - that sounds great! I'll keep an eye out.
Is this reasonable or too janky? https://github.com/great-expectations/great_expectations/compare/develop...edjoesu:feature/backend-args-for-diagnostics?expand=1

Because the execution engine params are common to all of the examples/tests, it is a bit awkward to put them in Expectation.examples, since it is a list, though it would be cleaner otherwise. 
Hi @edjoesu  - thanks for this! I need to have a think about the best way to do this, as you've raised some good points. A few questions/notes:
- Would you mind also updating a contrib expectation file to illustrate how this would work in practice? I'm having a hard time seeing how this propagates down to the individual SQL back-ends.
- You can see some examples of precedent in `tests/test_definitions`. We essentially have tests in JSON for each Expectation, which are parameterized and run using `test_expectations.py` and `test_expectations_cfe.py`. If you look, for instance, at `tests/test_definitions/column_map_expectations/expect_column_values_to_be_between.json`, you can see that there is an `only_for` key, which specifies which back-ends to run the test for. This system is not perfect, but I wonder if there is any inspiration to be had here.
- Would you mind opening this as a PR? Even if it's a bit rough, it will be easier to work with, and will be easier to share for internal discussion with the team.
I like the only_for implementation, but currently build_test_backends_list has to be called before iterating over the tests. 

It's possible we could have 'for_backends' replace only_for and take the union.

Commit message: Merge branch 'develop' into feature/backend-args-for-diagnostics

Patch to review:
diff --git a/contrib/experimental/great_expectations_experimental/expectations/expect_column_skew_to_be_between.py b/contrib/experimental/great_expectations_experimental/expectations/expect_column_skew_to_be_between.py
index c0dc1b1f6932..45d7bd7f5a6a 100644
--- a/contrib/experimental/great_expectations_experimental/expectations/expect_column_skew_to_be_between.py
+++ b/contrib/experimental/great_expectations_experimental/expectations/expect_column_skew_to_be_between.py
@@ -26,11 +26,13 @@
     ColumnMetricProvider,
     column_aggregate_value,
 )
-from great_expectations.expectations.metrics.import_manager import F, sa
-from great_expectations.expectations.metrics.metric_provider import (
-    MetricProvider,
-    metric_value,
+from great_expectations.expectations.metrics.column_aggregate_metric_provider import (
+    ColumnAggregateMetricProvider,
+    column_aggregate_partial,
+    column_aggregate_value,
 )
+from great_expectations.expectations.metrics.import_manager import F, sa
+from great_expectations.expectations.metrics.metric_provider import metric_value
 from great_expectations.expectations.util import render_evaluation_parameter_string
 from great_expectations.render.renderer.renderer import renderer
 from great_expectations.render.types import RenderedStringTemplateContent
@@ -55,34 +57,82 @@ def _pandas(cls, column, abs=False, **kwargs):
             return np.abs(stats.skew(column))
         return stats.skew(column)
 
-    #
-    # @metric_value(engine=SqlAlchemyExecutionEngine, metric_fn_type="value")
-    # def _sqlalchemy(
-    #     cls,
-    #     execution_engine: "SqlAlchemyExecutionEngine",
-    #     metric_domain_kwargs: Dict,
-    #     metric_value_kwargs: Dict,
-    #     metrics: Dict[Tuple, Any],
-    #     runtime_configuration: Dict,
-    # ):
-    #     (
-    #         selectable,
-    #         compute_domain_kwargs,
-    #         accessor_domain_kwargs,
-    #     ) = execution_engine.get_compute_domain(
-    #         metric_domain_kwargs, MetricDomainTypes.COLUMN
-    #     )
-    #     column_name = accessor_domain_kwargs["column"]
-    #     column = sa.column(column_name)
-    #     sqlalchemy_engine = execution_engine.engine
-    #     dialect = sqlalchemy_engine.dialect
-    #
-    #     column_median = None
-    #
-    #     # TODO: compute the value and return it
-    #
-    #     return column_median
-    #
+    @metric_value(engine=SqlAlchemyExecutionEngine)
+    def _sqlalchemy(
+        cls,
+        execution_engine: "SqlAlchemyExecutionEngine",
+        metric_domain_kwargs: Dict,
+        metric_value_kwargs: Dict,
+        metrics: Dict[Tuple, Any],
+        runtime_configuration: Dict,
+    ):
+        (
+            selectable,
+            compute_domain_kwargs,
+            accessor_domain_kwargs,
+        ) = execution_engine.get_compute_domain(
+            metric_domain_kwargs, MetricDomainTypes.COLUMN
+        )
+
+        column_name = accessor_domain_kwargs["column"]
+        column = sa.column(column_name)
+        sqlalchemy_engine = execution_engine.engine
+        dialect = sqlalchemy_engine.dialect
+
+        column_mean = _get_query_result(
+            func=sa.func.avg(column * 1.0),
+            selectable=selectable,
+            sqlalchemy_engine=sqlalchemy_engine,
+        )
+
+        column_count = _get_query_result(
+            func=sa.func.count(column),
+            selectable=selectable,
+            sqlalchemy_engine=sqlalchemy_engine,
+        )
+
+        if dialect.name.lower() == "mssql":
+            standard_deviation = sa.func.stdev(column)
+        else:
+            standard_deviation = sa.func.stddev_samp(column)
+
+        column_std = _get_query_result(
+            func=standard_deviation,
+            selectable=selectable,
+            sqlalchemy_engine=sqlalchemy_engine,
+        )
+
+        column_third_moment = _get_query_result(
+            func=sa.func.sum(sa.func.pow(column - column_mean, 3)),
+            selectable=selectable,
+            sqlalchemy_engine=sqlalchemy_engine,
+        )
+
+        column_skew = column_third_moment / (column_std ** 3) / (column_count - 1)
+        print(column_skew)
+        if metric_value_kwargs["abs"]:
+            return np.abs(column_skew)
+        else:
+            return column_skew
+
+
+def _get_query_result(func, selectable, sqlalchemy_engine):
+    simple_query: Select = sa.select(func).select_from(selectable)
+
+    try:
+        result: Row = sqlalchemy_engine.execute(simple_query).fetchone()[0]
+        print(result)
+        return result
+    except ProgrammingError as pe:
+        exception_message: str = "An SQL syntax Exception occurred."
+        exception_traceback: str = traceback.format_exc()
+        exception_message += (
+            f'{type(pe).__name__}: "{str(pe)}".  Traceback: "{exception_traceback}".'
+        )
+        logger.error(exception_message)
+        raise pe
+
+        #
     # @metric_value(engine=SparkDFExecutionEngine, metric_fn_type="value")
     # def _spark(
     #     cls,
@@ -229,6 +279,7 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "title": "positive_test_positive_skew",
                     "exact_match_out": False,
                     "include_in_gallery": True,
+                    "tolerance": 0.1,
                     "in": {"column": "a", "min_value": 0.25, "max_value": 10},
                     "out": {"success": True, "observed_value": 1.6974323016687487},
                 },
@@ -236,6 +287,7 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "title": "negative_test_no_skew",
                     "exact_match_out": False,
                     "include_in_gallery": True,
+                    "tolerance": 0.1,
                     "in": {"column": "b", "min_value": 0.25, "max_value": 10},
                     "out": {"success": False, "observed_value": -0.07638895580386174},
                 },
@@ -243,6 +295,7 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "title": "positive_test_negative_skew",
                     "exact_match_out": False,
                     "include_in_gallery": True,
+                    "tolerance": 0.1,
                     "in": {"column": "c", "min_value": -10, "max_value": -0.5},
                     "out": {"success": True, "observed_value": -0.9979514313860596},
                 },
@@ -250,6 +303,7 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "title": "negative_test_abs_skew",
                     "exact_match_out": False,
                     "include_in_gallery": True,
+                    "tolerance": 0.1,
                     "in": {
                         "column": "c",
                         "abs": True,
@@ -262,6 +316,7 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "title": "positive_test_abs_skew",
                     "exact_match_out": False,
                     "include_in_gallery": True,
+                    "tolerance": 0.1,
                     "in": {
                         "column": "c",
                         "abs": True,
@@ -271,7 +326,16 @@ class ExpectColumnSkewToBeBetween(ColumnExpectation):
                     "out": {"success": True, "observed_value": 0.9979514313860596},
                 },
             ],
-        },
+            "test_backends": [
+                {
+                    "backend": "pandas",
+                },
+                {
+                    "backend": "sqlalchemy",
+                    "dialects": ["sqlite", "postgresql"],
+                },
+            ],
+        }
     ]
 
     # This dictionary contains metadata for display in the public gallery
@@ -400,5 +464,7 @@ def _validate(
 
 
 if __name__ == "__main__":
-    self_check_report = ExpectColumnSkewToBeBetween().run_diagnostics()
+    self_check_report = ExpectColumnSkewToBeBetween().run_diagnostics(
+        execution_engine_args={"include_mysql": True}
+    )
     print(json.dumps(self_check_report, indent=2))
diff --git a/great_expectations/expectations/expectation.py b/great_expectations/expectations/expectation.py
index 6092c55c9411..288b213965a4 100644
--- a/great_expectations/expectations/expectation.py
+++ b/great_expectations/expectations/expectation.py
@@ -683,7 +683,7 @@ def build_configuration(cls, *args, **kwargs):
             meta=meta,
         )
 
-    def run_diagnostics(self, pretty_print=True):
+    def run_diagnostics(self, execution_engine_args=None, pretty_print=True):
         """
         Produce a diagnostic report about this expectation.
         The current uses for this method's output are
@@ -785,21 +785,21 @@ def run_diagnostics(self, pretty_print=True):
                     report_obj, e, traceback.format_exc()
                 )
 
-            execution_engines = None
+            introspected_execution_engines = None
             if upstream_metrics is not None:
-                execution_engines = self._get_execution_engine_dict(
+                introspected_execution_engines = self._get_execution_engine_dict(
                     upstream_metrics=upstream_metrics,
                 )
-                report_obj.update({"execution_engines": execution_engines})
+                report_obj.update({"execution_engines": introspected_execution_engines})
 
             try:
                 tests = self._get_examples(return_only_gallery_examples=False)
                 if len(tests) > 0:
-                    if execution_engines is not None:
+                    if introspected_execution_engines is not None:
                         test_results = self._get_test_results(
                             snake_name,
                             tests,
-                            execution_engines,
+                            introspected_execution_engines,
                         )
                         report_obj.update({"test_report": test_results})
             except Exception as e:
diff --git a/great_expectations/self_check/util.py b/great_expectations/self_check/util.py
index b1c44e51cec7..40819db86789 100644
--- a/great_expectations/self_check/util.py
+++ b/great_expectations/self_check/util.py
@@ -1470,27 +1470,46 @@ def generate_expectation_tests(
     parametrized_tests = []
 
     # use the expectation_execution_engines_dict (if provided) to request only the appropriate backends
-    if expectation_execution_engines_dict is not None:
-        backends = build_test_backends_list(
-            include_pandas=expectation_execution_engines_dict.get(
-                "PandasExecutionEngine"
-            )
-            == True,
-            include_spark=expectation_execution_engines_dict.get(
-                "SparkDFExecutionEngine"
-            )
-            == True,
-            include_sqlalchemy=expectation_execution_engines_dict.get(
-                "SqlAlchemyExecutionEngine"
+    for d in examples_config:
+        d = copy.deepcopy(d)
+        if expectation_execution_engines_dict is not None:
+            example_backends = [
+                backend_dict.get("backend")
+                for backend_dict in d.get("test_backends", [])
+            ]
+
+            example_sqlalchemy_dialects = [
+                dialect
+                for backend_dict in d.get("test_backends", {})
+                if (backend_dict.get("backend") == "sqlalchemy")
+                for dialect in backend_dict.get("dialects", [])
+            ]
+
+            backends = build_test_backends_list(
+                include_pandas=(
+                    expectation_execution_engines_dict.get("PandasExecutionEngine")
+                    == True
+                )
+                or ("pandas" in example_backends),
+                include_spark=(
+                    expectation_execution_engines_dict.get("SparkDFExecutionEngine")
+                    == True
+                )
+                or ("spark" in example_backends),
+                include_sqlalchemy=(
+                    expectation_execution_engines_dict.get("SqlAlchemyExecutionEngine")
+                    == True
+                )
+                or ("sqlalchemy" in example_backends),
+                include_postgresql=("postgresql" in example_sqlalchemy_dialects),
+                include_mysql=("mysql" in example_sqlalchemy_dialects),
+                include_mssql=("mssql" in example_sqlalchemy_dialects),
             )
-            == True,
-        )
-    else:
-        backends = build_test_backends_list()
+        else:
+            backends = build_test_backends_list()
+
+        for c in backends:
 
-    for c in backends:
-        for d in examples_config:
-            d = copy.deepcopy(d)
             datasets = []
             if candidate_test_is_on_temporary_notimplemented_list_cfe(
                 c, expectation_type
@@ -1854,6 +1873,7 @@ def check_json_test_result(test, result, data_asset=None):
         # NOTE - 20191031 - JPC - we may eventually want to change these tests as we update our view on how
         # representations, serializations, and objects should interact and how much of that is shown to the user.
         result = result.to_json_dict()
+        print(result)
         for key, value in test["out"].items():
             # Apply our great expectations-specific test logic
 


Code review comments:
diff_hunk: @@ -400,5 +464,7 @@ def _validate(
 
 
 if __name__ == "__main__":
-    self_check_report = ExpectColumnSkewToBeBetween().run_diagnostics()
+    self_check_report = ExpectColumnSkewToBeBetween().run_diagnostics(
+        execution_engine_args={"include_mysql": True}
path: contrib/experimental/great_expectations_experimental/expectations/expect_column_skew_to_be_between.py
text: @user1:
Would the `test_backends` dict above replace the need for the `execution_engine_args` parameter? Or are you providing two ways of specifying this?

@author:
Oops, left that in by mistake.