We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Validator throws KeyError 'table.head' while interactively creating Expectation Suite on a BigQuery datasource
**Describe the bug**
I am getting the same error as described in #3540 when interactively creating an Expectation Suite on a BigQuery datasource via CLI. As requested in the discussion, I am opening a new issue for this.

In the "Edit Your Expectation Suite" notebook provided by `great_expectations suite new`, the following function call throws an error:

```python
validator.head(n_rows=5, fetch_all=False)
```

Thrown error:

```text
KeyError                                  Traceback (most recent call last)
Input In [11], in <cell line: 1>()
----> 1 validator.head(n_rows=5, fetch_all=False)

File some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:2146, in Validator.head(self, n_rows, domain_kwargs, fetch_all)
   2141 if domain_kwargs is None:
   2142     domain_kwargs = {
   2143         "batch_id": self.execution_engine.active_batch_data_id,
   2144     }
-> 2146 data: Any = self.get_metric(
   2147     metric=MetricConfiguration(
   2148         metric_name="table.head",
   2149         metric_domain_kwargs=domain_kwargs,
   2150         metric_value_kwargs={
   2151             "n_rows": n_rows,
   2152             "fetch_all": fetch_all,
   2153         },
   2154     )
   2155 )
   2157 df: pd.DataFrame
   2158 if isinstance(
   2159     self.execution_engine, (PandasExecutionEngine, SqlAlchemyExecutionEngine)
   2160 ):

File some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:891, in Validator.get_metric(self, metric)
    889 def get_metric(self, metric: MetricConfiguration) -> Any:
    890     """return the value of the requested metric."""
--> 891     return self.get_metrics(metrics={metric.metric_name: metric})[
    892         metric.metric_name
    893     ]

File some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:856, in Validator.get_metrics(self, metrics)
    848 """
    849 metrics: Dictionary of desired metrics to be resolved, with metric_name as key and MetricConfiguration as value.
    850 Return Dictionary with requested metrics resolved, with metric_name as key and computed metric as value.
    851 """
    852 resolved_metrics: Dict[Tuple[str, str, str], Any] = self.compute_metrics(
    853     metric_configurations=list(metrics.values())
    854 )
--> 856 return {
    857     metric_configuration.metric_name: resolved_metrics[metric_configuration.id]
    858     for metric_configuration in metrics.values()
    859 }

File some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:857, in <dictcomp>(.0)
    848 """
    849 metrics: Dictionary of desired metrics to be resolved, with metric_name as key and MetricConfiguration as value.
    850 Return Dictionary with requested metrics resolved, with metric_name as key and computed metric as value.
    851 """
    852 resolved_metrics: Dict[Tuple[str, str, str], Any] = self.compute_metrics(
    853     metric_configurations=list(metrics.values())
    854 )
    856 return {
--> 857     metric_configuration.metric_name: resolved_metrics[metric_configuration.id]
    858     for metric_configuration in metrics.values()
    859 }

KeyError: ('table.head', 'batch_id=15a077d486452b3e1c894458758b7972', '04166707abe073177c1dd922d3584468')
```

**To Reproduce**
Steps to reproduce the behavior:

1. Initialize GE project
2. Add a BigQuery datasource via `great_expectations datasource new`
3. Create a new Expectation Suite via `great_expectations suite new`
4. Choose Interactively and select your datasource and data asset
5. Execute notebook code including the _validator.head()_ call
6. See error above

**Expected behavior**
Calling validator.head() should not raise a KeyError.

**Environment**
 - Operating System: MacOS 12.3.1
 - Great Expectations Version: 0.15.11

**Additional context**

I have examined the GCP logs in the period of the call of the validator.head() function. I exclude a permission error, because the used service account has maximum rights on used GCP project during debugging. However, errors occur here in the BigQuery service with the JobService.InsertJob method, which are not due to insufficient permissions:

```json
"serviceName": "bigquery.googleapis.com",
"methodName": "google.cloud.bigquery.v2.JobService.InsertJob",
"authorizationInfo": [
  {
    "resource": "projects/my-project",
    "permission": "bigquery.jobs.create",
    "granted": true,
    "resourceAttributes": {}
  }
],
```

The error itself is reported in the response object _jobStatus_:

```json
"jobStatus": {
  "errors": [
    {
      "code": 3,
      "message": "Cannot access field id on a value with type ARRAY<STRUCT<id STRING>> at [1:4656]"
    }
  ],
  "errorResult": {
    "message": "Cannot access field id on a value with type ARRAY<STRUCT<id STRING>> at [1:4656]",
    "code": 3
  },
  "jobState": "DONE"
},
```

Some fields of the table I use are nested fields. Does the validator have problems with these?

--- 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/table_metrics/table_head.py`
Content:
```
1 from typing import Any, Dict
2 
3 import pandas as pd
4 
5 from great_expectations.core.metric_domain_types import MetricDomainTypes
6 from great_expectations.execution_engine import (
7     PandasExecutionEngine,
8     SparkDFExecutionEngine,
9     SqlAlchemyExecutionEngine,
10 )
11 from great_expectations.expectations.metrics.import_manager import sa
12 from great_expectations.expectations.metrics.metric_provider import metric_value
13 from great_expectations.expectations.metrics.table_metric_provider import (
14     TableMetricProvider,
15 )
16 from great_expectations.validator.metric_configuration import MetricConfiguration
17 from great_expectations.validator.validator import Validator
18 
19 
20 class TableHead(TableMetricProvider):
21     metric_name = "table.head"
22     value_keys = ("n_rows", "fetch_all")
23     default_kwarg_values = {"n_rows": 5, "fetch_all": False}
24 
25     @metric_value(engine=PandasExecutionEngine)
26     def _pandas(
27         cls,
28         execution_engine: PandasExecutionEngine,
29         metric_domain_kwargs: Dict,
30         metric_value_kwargs: Dict,
31         metrics: Dict[str, Any],
32         runtime_configuration: Dict,
33     ):
34         df, _, _ = execution_engine.get_compute_domain(
35             metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE
36         )
37         if metric_value_kwargs.get("fetch_all", cls.default_kwarg_values["fetch_all"]):
38             return df
39         return df.head(metric_value_kwargs["n_rows"])
40 
41     @metric_value(engine=SqlAlchemyExecutionEngine)
42     def _sqlalchemy(
43         cls,
44         execution_engine: SqlAlchemyExecutionEngine,
45         metric_domain_kwargs: Dict,
46         metric_value_kwargs: Dict,
47         metrics: Dict[str, Any],
48         runtime_configuration: Dict,
49     ):
50         selectable, _, _ = execution_engine.get_compute_domain(
51             metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE
52         )
53         df = None
54         table_name = getattr(selectable, "name", None)
55         if table_name is None:
56             # if a custom query was passed
57             try:
58                 if metric_value_kwargs["fetch_all"]:
59                     df = pd.read_sql_query(
60                         sql=selectable,
61                         con=execution_engine.engine,
62                     )
63                 else:
64                     df = next(
65                         pd.read_sql_query(
66                             sql=selectable,
67                             con=execution_engine.engine,
68                             chunksize=metric_value_kwargs["n_rows"],
69                         )
70                     )
71             except (ValueError, NotImplementedError):
72                 # it looks like MetaData that is used by pd.read_sql_query
73                 # cannot work on a temp table.
74                 # If it fails, we are trying to get the data using read_sql
75                 df = None
76             except StopIteration:
77                 validator = Validator(execution_engine=execution_engine)
78                 columns = validator.get_metric(
79                     MetricConfiguration("table.columns", metric_domain_kwargs)
80                 )
81                 df = pd.DataFrame(columns=columns)
82         else:
83             try:
84                 if metric_value_kwargs["fetch_all"]:
85                     df = pd.read_sql_table(
86                         table_name=getattr(selectable, "name", None),
87                         schema=getattr(selectable, "schema", None),
88                         con=execution_engine.engine,
89                     )
90                 else:
91                     df = next(
92                         pd.read_sql_table(
93                             table_name=getattr(selectable, "name", None),
94                             schema=getattr(selectable, "schema", None),
95                             con=execution_engine.engine,
96                             chunksize=metric_value_kwargs["n_rows"],
97                         )
98                     )
99             except (ValueError, NotImplementedError):
100                 # it looks like MetaData that is used by pd.read_sql_table
101                 # cannot work on a temp table.
102                 # If it fails, we are trying to get the data using read_sql
103                 df = None
104             except StopIteration:
105                 validator = Validator(execution_engine=execution_engine)
106                 columns = validator.get_metric(
107                     MetricConfiguration("table.columns", metric_domain_kwargs)
108                 )
109                 df = pd.DataFrame(columns=columns)
110 
111         if df is None:
112             # we want to compile our selectable
113             stmt = sa.select(["*"]).select_from(selectable)
114             if metric_value_kwargs["fetch_all"]:
115                 sql = stmt.compile(
116                     dialect=execution_engine.engine.dialect,
117                     compile_kwargs={"literal_binds": True},
118                 )
119             elif execution_engine.engine.dialect.name.lower() == "mssql":
120                 # limit doesn't compile properly for mssql
121                 sql = str(
122                     stmt.compile(
123                         dialect=execution_engine.engine.dialect,
124                         compile_kwargs={"literal_binds": True},
125                     )
126                 )
127                 sql = f"SELECT TOP {metric_value_kwargs['n_rows']}{sql[6:]}"
128             else:
129                 stmt = stmt.limit(metric_value_kwargs["n_rows"])
130                 sql = stmt.compile(
131                     dialect=execution_engine.engine.dialect,
132                     compile_kwargs={"literal_binds": True},
133                 )
134 
135             df = pd.read_sql(sql, con=execution_engine.engine)
136 
137         return df
138 
139     @metric_value(engine=SparkDFExecutionEngine)
140     def _spark(
141         cls,
142         execution_engine: SparkDFExecutionEngine,
143         metric_domain_kwargs: Dict,
144         metric_value_kwargs: Dict,
145         metrics: Dict[str, Any],
146         runtime_configuration: Dict,
147     ):
148         df, _, _ = execution_engine.get_compute_domain(
149             metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE
150         )
151         if metric_value_kwargs["fetch_all"]:
152             return df.collect()
153         return df.head(metric_value_kwargs["n_rows"])
154
```
--- 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/table_metrics/table_head.py b/great_expectations/expectations/metrics/table_metrics/table_head.py
--- a/great_expectations/expectations/metrics/table_metrics/table_head.py
+++ b/great_expectations/expectations/metrics/table_metrics/table_head.py
@@ -52,7 +52,10 @@
         )
         df = None
         table_name = getattr(selectable, "name", None)
-        if table_name is None:
+        if (
+            isinstance(table_name, sa.sql.elements._anonymous_label)
+            or table_name is None
+        ):
             # if a custom query was passed
             try:
                 if metric_value_kwargs["fetch_all"]:

{"golden_diff": "diff --git a/great_expectations/expectations/metrics/table_metrics/table_head.py b/great_expectations/expectations/metrics/table_metrics/table_head.py\n--- a/great_expectations/expectations/metrics/table_metrics/table_head.py\n+++ b/great_expectations/expectations/metrics/table_metrics/table_head.py\n@@ -52,7 +52,10 @@\n         )\n         df = None\n         table_name = getattr(selectable, \"name\", None)\n-        if table_name is None:\n+        if (\n+            isinstance(table_name, sa.sql.elements._anonymous_label)\n+            or table_name is None\n+        ):\n             # if a custom query was passed\n             try:\n                 if metric_value_kwargs[\"fetch_all\"]:\n", "issue": "Validator throws KeyError 'table.head' while interactively creating Expectation Suite on a BigQuery datasource\n**Describe the bug**\r\nI am getting the same error as described in #3540 when interactively creating an Expectation Suite on a BigQuery datasource via CLI. As requested in the discussion, I am opening a new issue for this.\r\n\r\nIn the \"Edit Your Expectation Suite\" notebook provided by `great_expectations suite new`, the following function call throws an error:\r\n\r\n```python\r\nvalidator.head(n_rows=5, fetch_all=False)\r\n```\r\n\r\nThrown error:\r\n\r\n```text\r\nKeyError                                  Traceback (most recent call last)\r\nInput In [11], in <cell line: 1>()\r\n----> 1 validator.head(n_rows=5, fetch_all=False)\r\n\r\nFile some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:2146, in Validator.head(self, n_rows, domain_kwargs, fetch_all)\r\n   2141 if domain_kwargs is None:\r\n   2142     domain_kwargs = {\r\n   2143         \"batch_id\": self.execution_engine.active_batch_data_id,\r\n   2144     }\r\n-> 2146 data: Any = self.get_metric(\r\n   2147     metric=MetricConfiguration(\r\n   2148         metric_name=\"table.head\",\r\n   2149         metric_domain_kwargs=domain_kwargs,\r\n   2150         metric_value_kwargs={\r\n   2151             \"n_rows\": n_rows,\r\n   2152             \"fetch_all\": fetch_all,\r\n   2153         },\r\n   2154     )\r\n   2155 )\r\n   2157 df: pd.DataFrame\r\n   2158 if isinstance(\r\n   2159     self.execution_engine, (PandasExecutionEngine, SqlAlchemyExecutionEngine)\r\n   2160 ):\r\n\r\nFile some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:891, in Validator.get_metric(self, metric)\r\n    889 def get_metric(self, metric: MetricConfiguration) -> Any:\r\n    890     \"\"\"return the value of the requested metric.\"\"\"\r\n--> 891     return self.get_metrics(metrics={metric.metric_name: metric})[\r\n    892         metric.metric_name\r\n    893     ]\r\n\r\nFile some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:856, in Validator.get_metrics(self, metrics)\r\n    848 \"\"\"\r\n    849 metrics: Dictionary of desired metrics to be resolved, with metric_name as key and MetricConfiguration as value.\r\n    850 Return Dictionary with requested metrics resolved, with metric_name as key and computed metric as value.\r\n    851 \"\"\"\r\n    852 resolved_metrics: Dict[Tuple[str, str, str], Any] = self.compute_metrics(\r\n    853     metric_configurations=list(metrics.values())\r\n    854 )\r\n--> 856 return {\r\n    857     metric_configuration.metric_name: resolved_metrics[metric_configuration.id]\r\n    858     for metric_configuration in metrics.values()\r\n    859 }\r\n\r\nFile some-path/.venv/lib/python3.9/site-packages/great_expectations/validator/validator.py:857, in <dictcomp>(.0)\r\n    848 \"\"\"\r\n    849 metrics: Dictionary of desired metrics to be resolved, with metric_name as key and MetricConfiguration as value.\r\n    850 Return Dictionary with requested metrics resolved, with metric_name as key and computed metric as value.\r\n    851 \"\"\"\r\n    852 resolved_metrics: Dict[Tuple[str, str, str], Any] = self.compute_metrics(\r\n    853     metric_configurations=list(metrics.values())\r\n    854 )\r\n    856 return {\r\n--> 857     metric_configuration.metric_name: resolved_metrics[metric_configuration.id]\r\n    858     for metric_configuration in metrics.values()\r\n    859 }\r\n\r\nKeyError: ('table.head', 'batch_id=15a077d486452b3e1c894458758b7972', '04166707abe073177c1dd922d3584468')\r\n```\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior:\r\n\r\n1. Initialize GE project\r\n2. Add a BigQuery datasource via `great_expectations datasource new`\r\n3. Create a new Expectation Suite via `great_expectations suite new`\r\n4. Choose Interactively and select your datasource and data asset\r\n5. Execute notebook code including the _validator.head()_ call\r\n6. See error above\r\n\r\n**Expected behavior**\r\nCalling validator.head() should not raise a KeyError.\r\n\r\n**Environment**\r\n - Operating System: MacOS 12.3.1\r\n - Great Expectations Version: 0.15.11\r\n\r\n**Additional context**\r\n\r\nI have examined the GCP logs in the period of the call of the validator.head() function. I exclude a permission error, because the used service account has maximum rights on used GCP project during debugging. However, errors occur here in the BigQuery service with the JobService.InsertJob method, which are not due to insufficient permissions:\r\n\r\n```json\r\n\"serviceName\": \"bigquery.googleapis.com\",\r\n\"methodName\": \"google.cloud.bigquery.v2.JobService.InsertJob\",\r\n\"authorizationInfo\": [\r\n  {\r\n    \"resource\": \"projects/my-project\",\r\n    \"permission\": \"bigquery.jobs.create\",\r\n    \"granted\": true,\r\n    \"resourceAttributes\": {}\r\n  }\r\n],\r\n```\r\n\r\nThe error itself is reported in the response object _jobStatus_:\r\n\r\n```json\r\n\"jobStatus\": {\r\n  \"errors\": [\r\n    {\r\n      \"code\": 3,\r\n      \"message\": \"Cannot access field id on a value with type ARRAY<STRUCT<id STRING>> at [1:4656]\"\r\n    }\r\n  ],\r\n  \"errorResult\": {\r\n    \"message\": \"Cannot access field id on a value with type ARRAY<STRUCT<id STRING>> at [1:4656]\",\r\n    \"code\": 3\r\n  },\r\n  \"jobState\": \"DONE\"\r\n},\r\n```\r\n\r\nSome fields of the table I use are nested fields. Does the validator have problems with these?\n", "before_files": [{"content": "from typing import Any, Dict\n\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.import_manager import sa\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\nfrom great_expectations.expectations.metrics.table_metric_provider import (\n    TableMetricProvider,\n)\nfrom great_expectations.validator.metric_configuration import MetricConfiguration\nfrom great_expectations.validator.validator import Validator\n\n\nclass TableHead(TableMetricProvider):\n    metric_name = \"table.head\"\n    value_keys = (\"n_rows\", \"fetch_all\")\n    default_kwarg_values = {\"n_rows\": 5, \"fetch_all\": False}\n\n    @metric_value(engine=PandasExecutionEngine)\n    def _pandas(\n        cls,\n        execution_engine: PandasExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        df, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        if metric_value_kwargs.get(\"fetch_all\", cls.default_kwarg_values[\"fetch_all\"]):\n            return df\n        return df.head(metric_value_kwargs[\"n_rows\"])\n\n    @metric_value(engine=SqlAlchemyExecutionEngine)\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        selectable, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        df = None\n        table_name = getattr(selectable, \"name\", None)\n        if table_name is None:\n            # if a custom query was passed\n            try:\n                if metric_value_kwargs[\"fetch_all\"]:\n                    df = pd.read_sql_query(\n                        sql=selectable,\n                        con=execution_engine.engine,\n                    )\n                else:\n                    df = next(\n                        pd.read_sql_query(\n                            sql=selectable,\n                            con=execution_engine.engine,\n                            chunksize=metric_value_kwargs[\"n_rows\"],\n                        )\n                    )\n            except (ValueError, NotImplementedError):\n                # it looks like MetaData that is used by pd.read_sql_query\n                # cannot work on a temp table.\n                # If it fails, we are trying to get the data using read_sql\n                df = None\n            except StopIteration:\n                validator = Validator(execution_engine=execution_engine)\n                columns = validator.get_metric(\n                    MetricConfiguration(\"table.columns\", metric_domain_kwargs)\n                )\n                df = pd.DataFrame(columns=columns)\n        else:\n            try:\n                if metric_value_kwargs[\"fetch_all\"]:\n                    df = pd.read_sql_table(\n                        table_name=getattr(selectable, \"name\", None),\n                        schema=getattr(selectable, \"schema\", None),\n                        con=execution_engine.engine,\n                    )\n                else:\n                    df = next(\n                        pd.read_sql_table(\n                            table_name=getattr(selectable, \"name\", None),\n                            schema=getattr(selectable, \"schema\", None),\n                            con=execution_engine.engine,\n                            chunksize=metric_value_kwargs[\"n_rows\"],\n                        )\n                    )\n            except (ValueError, NotImplementedError):\n                # it looks like MetaData that is used by pd.read_sql_table\n                # cannot work on a temp table.\n                # If it fails, we are trying to get the data using read_sql\n                df = None\n            except StopIteration:\n                validator = Validator(execution_engine=execution_engine)\n                columns = validator.get_metric(\n                    MetricConfiguration(\"table.columns\", metric_domain_kwargs)\n                )\n                df = pd.DataFrame(columns=columns)\n\n        if df is None:\n            # we want to compile our selectable\n            stmt = sa.select([\"*\"]).select_from(selectable)\n            if metric_value_kwargs[\"fetch_all\"]:\n                sql = stmt.compile(\n                    dialect=execution_engine.engine.dialect,\n                    compile_kwargs={\"literal_binds\": True},\n                )\n            elif execution_engine.engine.dialect.name.lower() == \"mssql\":\n                # limit doesn't compile properly for mssql\n                sql = str(\n                    stmt.compile(\n                        dialect=execution_engine.engine.dialect,\n                        compile_kwargs={\"literal_binds\": True},\n                    )\n                )\n                sql = f\"SELECT TOP {metric_value_kwargs['n_rows']}{sql[6:]}\"\n            else:\n                stmt = stmt.limit(metric_value_kwargs[\"n_rows\"])\n                sql = stmt.compile(\n                    dialect=execution_engine.engine.dialect,\n                    compile_kwargs={\"literal_binds\": True},\n                )\n\n            df = pd.read_sql(sql, con=execution_engine.engine)\n\n        return df\n\n    @metric_value(engine=SparkDFExecutionEngine)\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        df, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        if metric_value_kwargs[\"fetch_all\"]:\n            return df.collect()\n        return df.head(metric_value_kwargs[\"n_rows\"])\n", "path": "great_expectations/expectations/metrics/table_metrics/table_head.py"}], "after_files": [{"content": "from typing import Any, Dict\n\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.import_manager import sa\nfrom great_expectations.expectations.metrics.metric_provider import metric_value\nfrom great_expectations.expectations.metrics.table_metric_provider import (\n    TableMetricProvider,\n)\nfrom great_expectations.validator.metric_configuration import MetricConfiguration\nfrom great_expectations.validator.validator import Validator\n\n\nclass TableHead(TableMetricProvider):\n    metric_name = \"table.head\"\n    value_keys = (\"n_rows\", \"fetch_all\")\n    default_kwarg_values = {\"n_rows\": 5, \"fetch_all\": False}\n\n    @metric_value(engine=PandasExecutionEngine)\n    def _pandas(\n        cls,\n        execution_engine: PandasExecutionEngine,\n        metric_domain_kwargs: Dict,\n        metric_value_kwargs: Dict,\n        metrics: Dict[str, Any],\n        runtime_configuration: Dict,\n    ):\n        df, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        if metric_value_kwargs.get(\"fetch_all\", cls.default_kwarg_values[\"fetch_all\"]):\n            return df\n        return df.head(metric_value_kwargs[\"n_rows\"])\n\n    @metric_value(engine=SqlAlchemyExecutionEngine)\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        selectable, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        df = None\n        table_name = getattr(selectable, \"name\", None)\n        if (\n            isinstance(table_name, sa.sql.elements._anonymous_label)\n            or table_name is None\n        ):\n            # if a custom query was passed\n            try:\n                if metric_value_kwargs[\"fetch_all\"]:\n                    df = pd.read_sql_query(\n                        sql=selectable,\n                        con=execution_engine.engine,\n                    )\n                else:\n                    df = next(\n                        pd.read_sql_query(\n                            sql=selectable,\n                            con=execution_engine.engine,\n                            chunksize=metric_value_kwargs[\"n_rows\"],\n                        )\n                    )\n            except (ValueError, NotImplementedError):\n                # it looks like MetaData that is used by pd.read_sql_query\n                # cannot work on a temp table.\n                # If it fails, we are trying to get the data using read_sql\n                df = None\n            except StopIteration:\n                validator = Validator(execution_engine=execution_engine)\n                columns = validator.get_metric(\n                    MetricConfiguration(\"table.columns\", metric_domain_kwargs)\n                )\n                df = pd.DataFrame(columns=columns)\n        else:\n            try:\n                if metric_value_kwargs[\"fetch_all\"]:\n                    df = pd.read_sql_table(\n                        table_name=getattr(selectable, \"name\", None),\n                        schema=getattr(selectable, \"schema\", None),\n                        con=execution_engine.engine,\n                    )\n                else:\n                    df = next(\n                        pd.read_sql_table(\n                            table_name=getattr(selectable, \"name\", None),\n                            schema=getattr(selectable, \"schema\", None),\n                            con=execution_engine.engine,\n                            chunksize=metric_value_kwargs[\"n_rows\"],\n                        )\n                    )\n            except (ValueError, NotImplementedError):\n                # it looks like MetaData that is used by pd.read_sql_table\n                # cannot work on a temp table.\n                # If it fails, we are trying to get the data using read_sql\n                df = None\n            except StopIteration:\n                validator = Validator(execution_engine=execution_engine)\n                columns = validator.get_metric(\n                    MetricConfiguration(\"table.columns\", metric_domain_kwargs)\n                )\n                df = pd.DataFrame(columns=columns)\n\n        if df is None:\n            # we want to compile our selectable\n            stmt = sa.select([\"*\"]).select_from(selectable)\n            if metric_value_kwargs[\"fetch_all\"]:\n                sql = stmt.compile(\n                    dialect=execution_engine.engine.dialect,\n                    compile_kwargs={\"literal_binds\": True},\n                )\n            elif execution_engine.engine.dialect.name.lower() == \"mssql\":\n                # limit doesn't compile properly for mssql\n                sql = str(\n                    stmt.compile(\n                        dialect=execution_engine.engine.dialect,\n                        compile_kwargs={\"literal_binds\": True},\n                    )\n                )\n                sql = f\"SELECT TOP {metric_value_kwargs['n_rows']}{sql[6:]}\"\n            else:\n                stmt = stmt.limit(metric_value_kwargs[\"n_rows\"])\n                sql = stmt.compile(\n                    dialect=execution_engine.engine.dialect,\n                    compile_kwargs={\"literal_binds\": True},\n                )\n\n            df = pd.read_sql(sql, con=execution_engine.engine)\n\n        return df\n\n    @metric_value(engine=SparkDFExecutionEngine)\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        df, _, _ = execution_engine.get_compute_domain(\n            metric_domain_kwargs, domain_type=MetricDomainTypes.TABLE\n        )\n        if metric_value_kwargs[\"fetch_all\"]:\n            return df.collect()\n        return df.head(metric_value_kwargs[\"n_rows\"])\n", "path": "great_expectations/expectations/metrics/table_metrics/table_head.py"}]}