We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Use cleaner solution for non-truncating division in python 2
Prefer `from __future__ import division` to `1.*x/y`

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py`
Content:
```
1 from typing import Any, Dict, Optional
2 
3 import numpy as np
4 import pandas as pd
5 
6 from great_expectations.core import ExpectationConfiguration
7 from great_expectations.core.metric_domain_types import MetricDomainTypes
8 from great_expectations.execution_engine import (
9     ExecutionEngine,
10     PandasExecutionEngine,
11     SparkDFExecutionEngine,
12     SqlAlchemyExecutionEngine,
13 )
14 from great_expectations.expectations.metrics.column_aggregate_metric_provider import (
15     ColumnAggregateMetricProvider,
16     column_aggregate_value,
17 )
18 from great_expectations.expectations.metrics.import_manager import sa
19 from great_expectations.expectations.metrics.metric_provider import metric_value
20 from great_expectations.validator.metric_configuration import MetricConfiguration
21 
22 
23 class ColumnMedian(ColumnAggregateMetricProvider):
24     """MetricProvider Class for Aggregate Mean MetricProvider"""
25 
26     metric_name = "column.median"
27 
28     @column_aggregate_value(engine=PandasExecutionEngine)
29     def _pandas(cls, column, **kwargs):
30         """Pandas Median Implementation"""
31         column_null_elements_cond: pd.Series = column.isnull()
32         column_nonnull_elements: pd.Series = column[~column_null_elements_cond]
33         return column_nonnull_elements.median()
34 
35     @metric_value(engine=SqlAlchemyExecutionEngine, metric_fn_type="value")
36     def _sqlalchemy(
37         cls,
38         execution_engine: SqlAlchemyExecutionEngine,
39         metric_domain_kwargs: Dict,
40         metric_value_kwargs: Dict,
41         metrics: Dict[str, Any],
42         runtime_configuration: Dict,
43     ):
44         (
45             selectable,
46             compute_domain_kwargs,
47             accessor_domain_kwargs,
48         ) = execution_engine.get_compute_domain(
49             metric_domain_kwargs, MetricDomainTypes.COLUMN
50         )
51         column_name = accessor_domain_kwargs["column"]
52         column = sa.column(column_name)
53         sqlalchemy_engine = execution_engine.engine
54         """SqlAlchemy Median Implementation"""
55         nonnull_count = metrics.get("column_values.nonnull.count")
56         if not nonnull_count:
57             return None
58 
59         element_values = sqlalchemy_engine.execute(
60             sa.select([column])
61             .order_by(column)
62             .where(column != None)
63             .offset(max(nonnull_count // 2 - 1, 0))
64             .limit(2)
65             .select_from(selectable)
66         )
67 
68         column_values = list(element_values.fetchall())
69 
70         if len(column_values) == 0:
71             column_median = None
72         elif nonnull_count % 2 == 0:
73             # An even number of column values: take the average of the two center values
74             column_median = (
75                 float(
76                     column_values[0][0]
77                     + column_values[1][0]  # left center value  # right center value
78                 )
79                 / 2.0
80             )  # Average center values
81         else:
82             # An odd number of column values, we can just take the center value
83             column_median = column_values[1][0]  # True center value
84         return column_median
85 
86     @metric_value(engine=SparkDFExecutionEngine, metric_fn_type="value")
87     def _spark(
88         cls,
89         execution_engine: SparkDFExecutionEngine,
90         metric_domain_kwargs: Dict,
91         metric_value_kwargs: Dict,
92         metrics: Dict[str, Any],
93         runtime_configuration: Dict,
94     ):
95         (
96             df,
97             compute_domain_kwargs,
98             accessor_domain_kwargs,
99         ) = execution_engine.get_compute_domain(
100             metric_domain_kwargs, MetricDomainTypes.COLUMN
101         )
102         column = accessor_domain_kwargs["column"]
103         # We will get the two middle values by choosing an epsilon to add
104         # to the 50th percentile such that we always get exactly the middle two values
105         # (i.e. 0 < epsilon < 1 / (2 * values))
106 
107         # Note that this can be an expensive computation; we are not exposing
108         # spark's ability to estimate.
109         # We add two to 2 * n_values to maintain a legitimate quantile
110         # in the degenerate case when n_values = 0
111 
112         """Spark Median Implementation"""
113         table_row_count = metrics.get("table.row_count")
114         result = df.approxQuantile(
115             column, [0.5, 0.5 + (1 / (2 + (2 * table_row_count)))], 0
116         )
117         return np.mean(result)
118 
119     @classmethod
120     def _get_evaluation_dependencies(
121         cls,
122         metric: MetricConfiguration,
123         configuration: Optional[ExpectationConfiguration] = None,
124         execution_engine: Optional[ExecutionEngine] = None,
125         runtime_configuration: Optional[dict] = None,
126     ):
127         """This should return a dictionary:
128         {
129           "dependency_name": MetricConfiguration,
130           ...
131         }
132         """
133         dependencies: dict = super()._get_evaluation_dependencies(
134             metric=metric,
135             configuration=configuration,
136             execution_engine=execution_engine,
137             runtime_configuration=runtime_configuration,
138         )
139 
140         if isinstance(execution_engine, SqlAlchemyExecutionEngine):
141             dependencies["column_values.nonnull.count"] = MetricConfiguration(
142                 metric_name="column_values.nonnull.count",
143                 metric_domain_kwargs=metric.metric_domain_kwargs,
144             )
145 
146         return dependencies
147
```

Path: `great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py`
Content:
```
1 from typing import Any, Dict, List, Optional
2 
3 import numpy as np
4 import pandas as pd
5 
6 from great_expectations.core.metric_domain_types import MetricDomainTypes
7 from great_expectations.execution_engine import (
8     PandasExecutionEngine,
9     SparkDFExecutionEngine,
10     SqlAlchemyExecutionEngine,
11 )
12 from great_expectations.expectations.metrics.column_aggregate_metric_provider import (
13     ColumnAggregateMetricProvider,
14 )
15 from great_expectations.expectations.metrics.import_manager import (
16     F,
17     pyspark_sql_DataFrame,
18     pyspark_sql_Row,
19     sa,
20     sa_sql_expression_Select,
21     sa_sql_expression_Selectable,
22     sqlalchemy_engine_Row,
23 )
24 from great_expectations.expectations.metrics.metric_provider import metric_value
25 
26 
27 class ColumnValueCounts(ColumnAggregateMetricProvider):
28     metric_name = "column.value_counts"
29     value_keys = ("sort", "collate")
30 
31     default_kwarg_values = {"sort": "value", "collate": None}
32 
33     @metric_value(engine=PandasExecutionEngine)
34     def _pandas(
35         cls,
36         execution_engine: PandasExecutionEngine,
37         metric_domain_kwargs: Dict[str, str],
38         metric_value_kwargs: Dict[str, Optional[str]],
39         **kwargs,
40     ) -> pd.Series:
41         sort: str = metric_value_kwargs.get("sort", cls.default_kwarg_values["sort"])
42         collate: Optional[str] = metric_value_kwargs.get(
43             "collate", cls.default_kwarg_values["collate"]
44         )
45 
46         if sort not in ["value", "count", "none"]:
47             raise ValueError("sort must be either 'value', 'count', or 'none'")
48         if collate is not None:
49             raise ValueError("collate parameter is not supported in PandasDataset")
50 
51         df: pd.DataFrame
52         accessor_domain_kwargs: Dict[str, str]
53         df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(
54             metric_domain_kwargs, MetricDomainTypes.COLUMN
55         )
56         column: str = accessor_domain_kwargs["column"]
57 
58         counts: pd.Series = df[column].value_counts()
59         if sort == "value":
60             try:
61                 counts.sort_index(inplace=True)
62             except TypeError:
63                 # Having values of multiple types in a object dtype column (e.g., strings and floats)
64                 # raises a TypeError when the sorting method performs comparisons.
65                 if df[column].dtype == object:
66                     counts.index = counts.index.astype(str)
67                     counts.sort_index(inplace=True)
68         elif sort == "counts":
69             counts.sort_values(inplace=True)
70         counts.name = "count"
71         counts.index.name = "value"
72         return counts
73 
74     @metric_value(engine=SqlAlchemyExecutionEngine)
75     def _sqlalchemy(
76         cls,
77         execution_engine: SqlAlchemyExecutionEngine,
78         metric_domain_kwargs: Dict[str, str],
79         metric_value_kwargs: Dict[str, Optional[str]],
80         **kwargs,
81     ) -> pd.Series:
82         sort: str = metric_value_kwargs.get("sort", cls.default_kwarg_values["sort"])
83         collate: Optional[str] = metric_value_kwargs.get(
84             "collate", cls.default_kwarg_values["collate"]
85         )
86 
87         if sort not in ["value", "count", "none"]:
88             raise ValueError("sort must be either 'value', 'count', or 'none'")
89         if collate is not None:
90             raise ValueError("collate parameter is not supported in PandasDataset")
91 
92         selectable: sa_sql_expression_Selectable
93         accessor_domain_kwargs: Dict[str, str]
94         selectable, _, accessor_domain_kwargs = execution_engine.get_compute_domain(
95             metric_domain_kwargs, MetricDomainTypes.COLUMN
96         )
97         column: str = accessor_domain_kwargs["column"]
98 
99         query: sa_sql_expression_Select = (
100             sa.select(
101                 [
102                     sa.column(column).label("value"),
103                     sa.func.count(sa.column(column)).label("count"),
104                 ]
105             )
106             .where(sa.column(column).is_not(None))
107             .group_by(sa.column(column))
108         )
109         if sort == "value":
110             # NOTE: depending on the way the underlying database collates columns,
111             # ordering can vary. postgresql collate "C" matches default sort
112             # for python and most other systems, but is not universally supported,
113             # so we use the default sort for the system, unless specifically overridden
114             if collate is not None:
115                 query = query.order_by(sa.column(column).collate(collate))
116             else:
117                 query = query.order_by(sa.column(column))
118         elif sort == "count":
119             query = query.order_by(sa.column("count").desc())
120         results: List[sqlalchemy_engine_Row] = execution_engine.engine.execute(
121             query.select_from(selectable)
122         ).fetchall()
123         # Numpy does not always infer the correct DataTypes for SqlAlchemy Row, so we cannot use vectorized approach.
124         series = pd.Series(
125             data=[row[1] for row in results],
126             index=pd.Index(data=[row[0] for row in results], name="value"),
127             name="count",
128             dtype="object",
129         )
130         return series
131 
132     @metric_value(engine=SparkDFExecutionEngine)
133     def _spark(
134         cls,
135         execution_engine: SparkDFExecutionEngine,
136         metric_domain_kwargs: Dict[str, str],
137         metric_value_kwargs: Dict[str, Optional[str]],
138         **kwargs,
139     ) -> pd.Series:
140         sort: str = metric_value_kwargs.get("sort", cls.default_kwarg_values["sort"])
141         collate: Optional[str] = metric_value_kwargs.get(
142             "collate", cls.default_kwarg_values["collate"]
143         )
144 
145         if sort not in ["value", "count", "none"]:
146             raise ValueError("sort must be either 'value', 'count', or 'none'")
147         if collate is not None:
148             raise ValueError("collate parameter is not supported in SparkDFDataset")
149 
150         df: pyspark_sql_DataFrame
151         accessor_domain_kwargs: Dict[str, str]
152         df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(
153             metric_domain_kwargs, MetricDomainTypes.COLUMN
154         )
155         column: str = accessor_domain_kwargs["column"]
156 
157         value_counts_df: pyspark_sql_DataFrame = (
158             df.select(column)
159             .where(F.col(column).isNotNull())
160             .where(~F.isnan(F.col(column)))
161             .groupBy(column)
162             .count()
163         )
164 
165         if sort == "value":
166             value_counts_df = value_counts_df.orderBy(column)
167         elif sort == "count":
168             value_counts_df = value_counts_df.orderBy(F.desc("count"))
169 
170         value_counts: List[pyspark_sql_Row] = value_counts_df.collect()
171 
172         # Numpy does not always infer the correct DataTypes for Spark df, so we cannot use vectorized approach.
173         values: List[Any]
174         counts: List[int]
175         if len(value_counts) > 0:
176             values, counts = zip(*value_counts)
177         else:
178             values = []
179             counts = []
180 
181         series = pd.Series(
182             counts,
183             index=pd.Index(data=values, name="value"),
184             name="count",
185         )
186         return series
187
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py b/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py
--- a/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py
+++ b/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py
@@ -80,7 +80,11 @@
             )  # Average center values
         else:
             # An odd number of column values, we can just take the center value
-            column_median = column_values[1][0]  # True center value
+            if len(column_values) == 1:
+                column_median = column_values[0][0]  # The only value
+            else:
+                column_median = column_values[1][0]  # True center value
+
         return column_median
 
     @metric_value(engine=SparkDFExecutionEngine, metric_fn_type="value")
diff --git a/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py b/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py
--- a/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py
+++ b/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py
@@ -155,11 +155,7 @@
         column: str = accessor_domain_kwargs["column"]
 
         value_counts_df: pyspark_sql_DataFrame = (
-            df.select(column)
-            .where(F.col(column).isNotNull())
-            .where(~F.isnan(F.col(column)))
-            .groupBy(column)
-            .count()
+            df.select(column).where(F.col(column).isNotNull()).groupBy(column).count()
         )
 
         if sort == "value":

{"golden_diff": "diff --git a/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py b/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py\n--- a/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py\n+++ b/great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py\n@@ -80,7 +80,11 @@\n             )  # Average center values\n         else:\n             # An odd number of column values, we can just take the center value\n-            column_median = column_values[1][0]  # True center value\n+            if len(column_values) == 1:\n+                column_median = column_values[0][0]  # The only value\n+            else:\n+                column_median = column_values[1][0]  # True center value\n+\n         return column_median\n \n     @metric_value(engine=SparkDFExecutionEngine, metric_fn_type=\"value\")\ndiff --git a/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py b/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py\n--- a/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py\n+++ b/great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py\n@@ -155,11 +155,7 @@\n         column: str = accessor_domain_kwargs[\"column\"]\n \n         value_counts_df: pyspark_sql_DataFrame = (\n-            df.select(column)\n-            .where(F.col(column).isNotNull())\n-            .where(~F.isnan(F.col(column)))\n-            .groupBy(column)\n-            .count()\n+            df.select(column).where(F.col(column).isNotNull()).groupBy(column).count()\n         )\n \n         if sort == \"value\":\n", "issue": "Use cleaner solution for non-truncating division in python 2\nPrefer `from __future__ import division` to `1.*x/y`\n", "before_files": [{"content": "from typing import Any, Dict, Optional\n\nimport numpy as np\nimport pandas as pd\n\nfrom great_expectations.core import ExpectationConfiguration\nfrom great_expectations.core.metric_domain_types import MetricDomainTypes\nfrom great_expectations.execution_engine import (\n    ExecutionEngine,\n    PandasExecutionEngine,\n    SparkDFExecutionEngine,\n    SqlAlchemyExecutionEngine,\n)\nfrom great_expectations.expectations.metrics.column_aggregate_metric_provider import (\n    ColumnAggregateMetricProvider,\n    column_aggregate_value,\n)\nfrom great_expectations.expectations.metrics.import_manager import sa\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\nfrom great_expectations.validator.metric_configuration import MetricConfiguration\n\n\nclass ColumnMedian(ColumnAggregateMetricProvider):\n    \"\"\"MetricProvider Class for Aggregate Mean MetricProvider\"\"\"\n\n    metric_name = \"column.median\"\n\n    @column_aggregate_value(engine=PandasExecutionEngine)\n    def _pandas(cls, column, **kwargs):\n        \"\"\"Pandas Median Implementation\"\"\"\n        column_null_elements_cond: pd.Series = column.isnull()\n        column_nonnull_elements: pd.Series = column[~column_null_elements_cond]\n        return column_nonnull_elements.median()\n\n    @metric_value(engine=SqlAlchemyExecutionEngine, metric_fn_type=\"value\")\n    def _sqlalchemy(\n        cls,\n        execution_engine: SqlAlchemyExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        (\n            selectable,\n            compute_domain_kwargs,\n            accessor_domain_kwargs,\n        ) = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column_name = accessor_domain_kwargs[\"column\"]\n        column = sa.column(column_name)\n        sqlalchemy_engine = execution_engine.engine\n        \"\"\"SqlAlchemy Median Implementation\"\"\"\n        nonnull_count = metrics.get(\"column_values.nonnull.count\")\n        if not nonnull_count:\n            return None\n\n        element_values = sqlalchemy_engine.execute(\n            sa.select([column])\n            .order_by(column)\n            .where(column != None)\n            .offset(max(nonnull_count // 2 - 1, 0))\n            .limit(2)\n            .select_from(selectable)\n        )\n\n        column_values = list(element_values.fetchall())\n\n        if len(column_values) == 0:\n            column_median = None\n        elif nonnull_count % 2 == 0:\n            # An even number of column values: take the average of the two center values\n            column_median = (\n                float(\n                    column_values[0][0]\n                    + column_values[1][0]  # left center value  # right center value\n                )\n                / 2.0\n            )  # Average center values\n        else:\n            # An odd number of column values, we can just take the center value\n            column_median = column_values[1][0]  # True center value\n        return column_median\n\n    @metric_value(engine=SparkDFExecutionEngine, metric_fn_type=\"value\")\n    def _spark(\n        cls,\n        execution_engine: SparkDFExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        (\n            df,\n            compute_domain_kwargs,\n            accessor_domain_kwargs,\n        ) = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column = accessor_domain_kwargs[\"column\"]\n        # We will get the two middle values by choosing an epsilon to add\n        # to the 50th percentile such that we always get exactly the middle two values\n        # (i.e. 0 < epsilon < 1 / (2 * values))\n\n        # Note that this can be an expensive computation; we are not exposing\n        # spark's ability to estimate.\n        # We add two to 2 * n_values to maintain a legitimate quantile\n        # in the degenerate case when n_values = 0\n\n        \"\"\"Spark Median Implementation\"\"\"\n        table_row_count = metrics.get(\"table.row_count\")\n        result = df.approxQuantile(\n            column, [0.5, 0.5 + (1 / (2 + (2 * table_row_count)))], 0\n        )\n        return np.mean(result)\n\n    @classmethod\n    def _get_evaluation_dependencies(\n        cls,\n        metric: MetricConfiguration,\n        configuration: Optional[ExpectationConfiguration] = None,\n        execution_engine: Optional[ExecutionEngine] = None,\n        runtime_configuration: Optional[dict] = None,\n    ):\n        \"\"\"This should return a dictionary:\n        {\n          \"dependency_name\": MetricConfiguration,\n          ...\n        }\n        \"\"\"\n        dependencies: dict = super()._get_evaluation_dependencies(\n            metric=metric,\n            configuration=configuration,\n            execution_engine=execution_engine,\n            runtime_configuration=runtime_configuration,\n        )\n\n        if isinstance(execution_engine, SqlAlchemyExecutionEngine):\n            dependencies[\"column_values.nonnull.count\"] = MetricConfiguration(\n                metric_name=\"column_values.nonnull.count\",\n                metric_domain_kwargs=metric.metric_domain_kwargs,\n            )\n\n        return dependencies\n", "path": "great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py"}, {"content": "from typing import Any, Dict, List, Optional\n\nimport numpy as np\nimport pandas as pd\n\nfrom great_expectations.core.metric_domain_types import MetricDomainTypes\nfrom great_expectations.execution_engine import (\n    PandasExecutionEngine,\n    SparkDFExecutionEngine,\n    SqlAlchemyExecutionEngine,\n)\nfrom great_expectations.expectations.metrics.column_aggregate_metric_provider import (\n    ColumnAggregateMetricProvider,\n)\nfrom great_expectations.expectations.metrics.import_manager import (\n    F,\n    pyspark_sql_DataFrame,\n    pyspark_sql_Row,\n    sa,\n    sa_sql_expression_Select,\n    sa_sql_expression_Selectable,\n    sqlalchemy_engine_Row,\n)\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\n\n\nclass ColumnValueCounts(ColumnAggregateMetricProvider):\n    metric_name = \"column.value_counts\"\n    value_keys = (\"sort\", \"collate\")\n\n    default_kwarg_values = {\"sort\": \"value\", \"collate\": None}\n\n    @metric_value(engine=PandasExecutionEngine)\n    def _pandas(\n        cls,\n        execution_engine: PandasExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in PandasDataset\")\n\n        df: pd.DataFrame\n        accessor_domain_kwargs: Dict[str, str]\n        df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        counts: pd.Series = df[column].value_counts()\n        if sort == \"value\":\n            try:\n                counts.sort_index(inplace=True)\n            except TypeError:\n                # Having values of multiple types in a object dtype column (e.g., strings and floats)\n                # raises a TypeError when the sorting method performs comparisons.\n                if df[column].dtype == object:\n                    counts.index = counts.index.astype(str)\n                    counts.sort_index(inplace=True)\n        elif sort == \"counts\":\n            counts.sort_values(inplace=True)\n        counts.name = \"count\"\n        counts.index.name = \"value\"\n        return counts\n\n    @metric_value(engine=SqlAlchemyExecutionEngine)\n    def _sqlalchemy(\n        cls,\n        execution_engine: SqlAlchemyExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in PandasDataset\")\n\n        selectable: sa_sql_expression_Selectable\n        accessor_domain_kwargs: Dict[str, str]\n        selectable, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        query: sa_sql_expression_Select = (\n            sa.select(\n                [\n                    sa.column(column).label(\"value\"),\n                    sa.func.count(sa.column(column)).label(\"count\"),\n                ]\n            )\n            .where(sa.column(column).is_not(None))\n            .group_by(sa.column(column))\n        )\n        if sort == \"value\":\n            # NOTE: depending on the way the underlying database collates columns,\n            # ordering can vary. postgresql collate \"C\" matches default sort\n            # for python and most other systems, but is not universally supported,\n            # so we use the default sort for the system, unless specifically overridden\n            if collate is not None:\n                query = query.order_by(sa.column(column).collate(collate))\n            else:\n                query = query.order_by(sa.column(column))\n        elif sort == \"count\":\n            query = query.order_by(sa.column(\"count\").desc())\n        results: List[sqlalchemy_engine_Row] = execution_engine.engine.execute(\n            query.select_from(selectable)\n        ).fetchall()\n        # Numpy does not always infer the correct DataTypes for SqlAlchemy Row, so we cannot use vectorized approach.\n        series = pd.Series(\n            data=[row[1] for row in results],\n            index=pd.Index(data=[row[0] for row in results], name=\"value\"),\n            name=\"count\",\n            dtype=\"object\",\n        )\n        return series\n\n    @metric_value(engine=SparkDFExecutionEngine)\n    def _spark(\n        cls,\n        execution_engine: SparkDFExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in SparkDFDataset\")\n\n        df: pyspark_sql_DataFrame\n        accessor_domain_kwargs: Dict[str, str]\n        df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        value_counts_df: pyspark_sql_DataFrame = (\n            df.select(column)\n            .where(F.col(column).isNotNull())\n            .where(~F.isnan(F.col(column)))\n            .groupBy(column)\n            .count()\n        )\n\n        if sort == \"value\":\n            value_counts_df = value_counts_df.orderBy(column)\n        elif sort == \"count\":\n            value_counts_df = value_counts_df.orderBy(F.desc(\"count\"))\n\n        value_counts: List[pyspark_sql_Row] = value_counts_df.collect()\n\n        # Numpy does not always infer the correct DataTypes for Spark df, so we cannot use vectorized approach.\n        values: List[Any]\n        counts: List[int]\n        if len(value_counts) > 0:\n            values, counts = zip(*value_counts)\n        else:\n            values = []\n            counts = []\n\n        series = pd.Series(\n            counts,\n            index=pd.Index(data=values, name=\"value\"),\n            name=\"count\",\n        )\n        return series\n", "path": "great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py"}], "after_files": [{"content": "from typing import Any, Dict, Optional\n\nimport numpy as np\nimport pandas as pd\n\nfrom great_expectations.core import ExpectationConfiguration\nfrom great_expectations.core.metric_domain_types import MetricDomainTypes\nfrom great_expectations.execution_engine import (\n    ExecutionEngine,\n    PandasExecutionEngine,\n    SparkDFExecutionEngine,\n    SqlAlchemyExecutionEngine,\n)\nfrom great_expectations.expectations.metrics.column_aggregate_metric_provider import (\n    ColumnAggregateMetricProvider,\n    column_aggregate_value,\n)\nfrom great_expectations.expectations.metrics.import_manager import sa\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\nfrom great_expectations.validator.metric_configuration import MetricConfiguration\n\n\nclass ColumnMedian(ColumnAggregateMetricProvider):\n    \"\"\"MetricProvider Class for Aggregate Mean MetricProvider\"\"\"\n\n    metric_name = \"column.median\"\n\n    @column_aggregate_value(engine=PandasExecutionEngine)\n    def _pandas(cls, column, **kwargs):\n        \"\"\"Pandas Median Implementation\"\"\"\n        column_null_elements_cond: pd.Series = column.isnull()\n        column_nonnull_elements: pd.Series = column[~column_null_elements_cond]\n        return column_nonnull_elements.median()\n\n    @metric_value(engine=SqlAlchemyExecutionEngine, metric_fn_type=\"value\")\n    def _sqlalchemy(\n        cls,\n        execution_engine: SqlAlchemyExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        (\n            selectable,\n            compute_domain_kwargs,\n            accessor_domain_kwargs,\n        ) = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column_name = accessor_domain_kwargs[\"column\"]\n        column = sa.column(column_name)\n        sqlalchemy_engine = execution_engine.engine\n        \"\"\"SqlAlchemy Median Implementation\"\"\"\n        nonnull_count = metrics.get(\"column_values.nonnull.count\")\n        if not nonnull_count:\n            return None\n\n        element_values = sqlalchemy_engine.execute(\n            sa.select([column])\n            .order_by(column)\n            .where(column != None)\n            .offset(max(nonnull_count // 2 - 1, 0))\n            .limit(2)\n            .select_from(selectable)\n        )\n\n        column_values = list(element_values.fetchall())\n\n        if len(column_values) == 0:\n            column_median = None\n        elif nonnull_count % 2 == 0:\n            # An even number of column values: take the average of the two center values\n            column_median = (\n                float(\n                    column_values[0][0]\n                    + column_values[1][0]  # left center value  # right center value\n                )\n                / 2.0\n            )  # Average center values\n        else:\n            # An odd number of column values, we can just take the center value\n            if len(column_values) == 1:\n                column_median = column_values[0][0]  # The only value\n            else:\n                column_median = column_values[1][0]  # True center value\n\n        return column_median\n\n    @metric_value(engine=SparkDFExecutionEngine, metric_fn_type=\"value\")\n    def _spark(\n        cls,\n        execution_engine: SparkDFExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        (\n            df,\n            compute_domain_kwargs,\n            accessor_domain_kwargs,\n        ) = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column = accessor_domain_kwargs[\"column\"]\n        # We will get the two middle values by choosing an epsilon to add\n        # to the 50th percentile such that we always get exactly the middle two values\n        # (i.e. 0 < epsilon < 1 / (2 * values))\n\n        # Note that this can be an expensive computation; we are not exposing\n        # spark's ability to estimate.\n        # We add two to 2 * n_values to maintain a legitimate quantile\n        # in the degenerate case when n_values = 0\n\n        \"\"\"Spark Median Implementation\"\"\"\n        table_row_count = metrics.get(\"table.row_count\")\n        result = df.approxQuantile(\n            column, [0.5, 0.5 + (1 / (2 + (2 * table_row_count)))], 0\n        )\n        return np.mean(result)\n\n    @classmethod\n    def _get_evaluation_dependencies(\n        cls,\n        metric: MetricConfiguration,\n        configuration: Optional[ExpectationConfiguration] = None,\n        execution_engine: Optional[ExecutionEngine] = None,\n        runtime_configuration: Optional[dict] = None,\n    ):\n        \"\"\"This should return a dictionary:\n        {\n          \"dependency_name\": MetricConfiguration,\n          ...\n        }\n        \"\"\"\n        dependencies: dict = super()._get_evaluation_dependencies(\n            metric=metric,\n            configuration=configuration,\n            execution_engine=execution_engine,\n            runtime_configuration=runtime_configuration,\n        )\n\n        if isinstance(execution_engine, SqlAlchemyExecutionEngine):\n            dependencies[\"column_values.nonnull.count\"] = MetricConfiguration(\n                metric_name=\"column_values.nonnull.count\",\n                metric_domain_kwargs=metric.metric_domain_kwargs,\n            )\n\n        return dependencies\n", "path": "great_expectations/expectations/metrics/column_aggregate_metrics/column_median.py"}, {"content": "from typing import Any, Dict, List, Optional\n\nimport numpy as np\nimport pandas as pd\n\nfrom great_expectations.core.metric_domain_types import MetricDomainTypes\nfrom great_expectations.execution_engine import (\n    PandasExecutionEngine,\n    SparkDFExecutionEngine,\n    SqlAlchemyExecutionEngine,\n)\nfrom great_expectations.expectations.metrics.column_aggregate_metric_provider import (\n    ColumnAggregateMetricProvider,\n)\nfrom great_expectations.expectations.metrics.import_manager import (\n    F,\n    pyspark_sql_DataFrame,\n    pyspark_sql_Row,\n    sa,\n    sa_sql_expression_Select,\n    sa_sql_expression_Selectable,\n    sqlalchemy_engine_Row,\n)\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\n\n\nclass ColumnValueCounts(ColumnAggregateMetricProvider):\n    metric_name = \"column.value_counts\"\n    value_keys = (\"sort\", \"collate\")\n\n    default_kwarg_values = {\"sort\": \"value\", \"collate\": None}\n\n    @metric_value(engine=PandasExecutionEngine)\n    def _pandas(\n        cls,\n        execution_engine: PandasExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in PandasDataset\")\n\n        df: pd.DataFrame\n        accessor_domain_kwargs: Dict[str, str]\n        df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        counts: pd.Series = df[column].value_counts()\n        if sort == \"value\":\n            try:\n                counts.sort_index(inplace=True)\n            except TypeError:\n                # Having values of multiple types in a object dtype column (e.g., strings and floats)\n                # raises a TypeError when the sorting method performs comparisons.\n                if df[column].dtype == object:\n                    counts.index = counts.index.astype(str)\n                    counts.sort_index(inplace=True)\n        elif sort == \"counts\":\n            counts.sort_values(inplace=True)\n        counts.name = \"count\"\n        counts.index.name = \"value\"\n        return counts\n\n    @metric_value(engine=SqlAlchemyExecutionEngine)\n    def _sqlalchemy(\n        cls,\n        execution_engine: SqlAlchemyExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in PandasDataset\")\n\n        selectable: sa_sql_expression_Selectable\n        accessor_domain_kwargs: Dict[str, str]\n        selectable, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        query: sa_sql_expression_Select = (\n            sa.select(\n                [\n                    sa.column(column).label(\"value\"),\n                    sa.func.count(sa.column(column)).label(\"count\"),\n                ]\n            )\n            .where(sa.column(column).is_not(None))\n            .group_by(sa.column(column))\n        )\n        if sort == \"value\":\n            # NOTE: depending on the way the underlying database collates columns,\n            # ordering can vary. postgresql collate \"C\" matches default sort\n            # for python and most other systems, but is not universally supported,\n            # so we use the default sort for the system, unless specifically overridden\n            if collate is not None:\n                query = query.order_by(sa.column(column).collate(collate))\n            else:\n                query = query.order_by(sa.column(column))\n        elif sort == \"count\":\n            query = query.order_by(sa.column(\"count\").desc())\n        results: List[sqlalchemy_engine_Row] = execution_engine.engine.execute(\n            query.select_from(selectable)\n        ).fetchall()\n        # Numpy does not always infer the correct DataTypes for SqlAlchemy Row, so we cannot use vectorized approach.\n        series = pd.Series(\n            data=[row[1] for row in results],\n            index=pd.Index(data=[row[0] for row in results], name=\"value\"),\n            name=\"count\",\n            dtype=\"object\",\n        )\n        return series\n\n    @metric_value(engine=SparkDFExecutionEngine)\n    def _spark(\n        cls,\n        execution_engine: SparkDFExecutionEngine,\n        metric_domain_kwargs: Dict[str, str],\n        metric_value_kwargs: Dict[str, Optional[str]],\n        **kwargs,\n    ) -> pd.Series:\n        sort: str = metric_value_kwargs.get(\"sort\", cls.default_kwarg_values[\"sort\"])\n        collate: Optional[str] = metric_value_kwargs.get(\n            \"collate\", cls.default_kwarg_values[\"collate\"]\n        )\n\n        if sort not in [\"value\", \"count\", \"none\"]:\n            raise ValueError(\"sort must be either 'value', 'count', or 'none'\")\n        if collate is not None:\n            raise ValueError(\"collate parameter is not supported in SparkDFDataset\")\n\n        df: pyspark_sql_DataFrame\n        accessor_domain_kwargs: Dict[str, str]\n        df, _, accessor_domain_kwargs = execution_engine.get_compute_domain(\n            metric_domain_kwargs, MetricDomainTypes.COLUMN\n        )\n        column: str = accessor_domain_kwargs[\"column\"]\n\n        value_counts_df: pyspark_sql_DataFrame = (\n            df.select(column).where(F.col(column).isNotNull()).groupBy(column).count()\n        )\n\n        if sort == \"value\":\n            value_counts_df = value_counts_df.orderBy(column)\n        elif sort == \"count\":\n            value_counts_df = value_counts_df.orderBy(F.desc(\"count\"))\n\n        value_counts: List[pyspark_sql_Row] = value_counts_df.collect()\n\n        # Numpy does not always infer the correct DataTypes for Spark df, so we cannot use vectorized approach.\n        values: List[Any]\n        counts: List[int]\n        if len(value_counts) > 0:\n            values, counts = zip(*value_counts)\n        else:\n            values = []\n            counts = []\n\n        series = pd.Series(\n            counts,\n            index=pd.Index(data=values, name=\"value\"),\n            name=\"count\",\n        )\n        return series\n", "path": "great_expectations/expectations/metrics/column_aggregate_metrics/column_value_counts.py"}]}