Repository: great-expectations/great_expectations
Pull request: [FEATURE] Support to include ID/PK in validation result for each row - Spark
### Changes proposed in this pull request:

PySpark implementation of `unexpected_index_column`s value that allows users to specify a primary key (PK) column for identifying rows that failed an Expectation (usually returned as part of the `unexpected_index_list`).  The PR also enables `unexpected_index_query` to be returned to the user, which provides a way for the user to retrieve the full list of unexpected results from the PySpark DataFrame.

Changes were made to the `map_metric_provider` to take in the parameter from `result_format` and output `unexpected_index_list` as key-value pairs of the primary key column.

Note : This is only the Spark implementation, [Pandas has been merged already](https://github.com/great-expectations/great_expectations/pull/6329), and [Spark has been merged as well](https://github.com/great-expectations/great_expectations/pull/6448).

* closes #3195

### What has changed?
* `unexpected_index_list` and `unexpected_index_query` were added as validation dependencies for `SparkDExecutionEngine`
* `_spark_map_condition_index` was added as MapMetric.
   - This Metric returns the indices that contain unexpected values, according to the Primary Key values specified using `unexpected_index_column_names`.
* `_spark_map_condition_index` was added as MapMetric.
   - This Metric returns the indices that contain unexpected values, according to the Primary Key values specified using `unexpected_index_column_names`.
   - Like PR #6660 the MapMetric also supports the `return_unexpected_index_query` flag to suppress output if desired.

### Can you give me an Example? (heavily adopted from the [Pandas](https://github.com/great-expectations/great_expectations/pull/6329) and [SQL](https://github.com/great-expectations/great_expectations/pull/6448) Examples)

Given the following PySpark DataFrame
```python 
 df: pd.DataFrame = pd.DataFrame(
        {
            "pk_1": [0, 1, 2, 3, 4, 5],
            "pk_2": ["zero", "one", "two", "three", "four", "five"],
            "animals": [
                "cat",
                "fish",
                "dog",
                "giraffe",
                "lion",
                "zebra",
            ],
        }
    )
    test_df: "pyspark.sql.dataframe.DataFrame" = spark_session.createDataFrame(
        data=df,
    )  # noqa: F821
```

We could run the `expect_column_values_to_be_in_set` Expectation on the `animals` column  with  `["cat", "fish", "dog"]` as the `value_set` (ie domestic animals).

```python
expectationConfiguration = ExpectationConfiguration(
    expectation_type="expect_column_values_to_be_in_set",
    kwargs={
        "column": "animals",
        "value_set": ["cat", "fish", "dog"],
    },
)
```
We could run a `Checkpoint` that contains the `ExpectationConfiguration` with the following `result_format` configuration, which sets `unexpected_index_columns` to be `pk_1`: 

```python
result_format: dict = {
        "result_format": "COMPLETE",
        "unexpected_index_column_names": ["pk_1"],
    }
```

After running the `Checkpoint`, we expect the validation result to contain the following `unexpected_index_list`: 
```python
[{"animals": "giraffe", "pk_1": 3},
{"animals": "lion", "pk_1": 4},
{"animals": "zebra", "pk_1": 5}]
```

#### What if I have a **lot** of rows in my table?

In order to retrieve the *full list* unexpected values from the table, the result also contains a `unexpected_index_query`, which can be used against the original date (represented by `df`), to retrieve the full set of unexpected results. 
 
```python
df.filter(F.expr((animals IS NOT NULL) AND (NOT (animals IN (cat, fish, dog))))
```
The query (along with `unexpected_index_list`) is returned as part of the result values returned by the `Validator`. 

```python 
...
"unexpected_index_list": [{"pk_1": 3}, {"pk_1": 4}, {"pk_1": 5}],
"unexpected_index_query": "df.filter(F.expr((animals IS NOT NULL) AND (NOT (animals IN (cat, fish, dog))))"
...
```

### Definition of Done
Please delete options that are not relevant.

- [x] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [x] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_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 added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [x] I have run any local integration tests and made sure that nothing is broken.

Problem statement:
Support to include ID/PK in validation result for each row that failed an expectations
**Is your feature request related to a problem? Please describe.**
When an expectation is run the output includes a “partial_unexpected_list” property of values that were unexpected. While this is useful, in most cases, it doesn't allow teams to identify, resolve, or divert data with poor quality.

It would be great if a sample list or a complete list (result_format SUMMARY, COMPLETE) of ID's for each row that failed an expectation was included in the validation result.

**Describe the solution you'd like**
One could include a column name as an argument in the expectation that they would like to be included in the "partial_unexpected_list" (or a new property).

**Describe alternatives you've considered**
One could try infer the PK for a table but this is not possible for all engine types E.g. Spark. 

**Additional context**
Enabling teams to not only bring light to data quality issues but identify all rows allows them to address poor data quality in real-time instead of requiring manual intervention.

Hints from issue discussion:
This is a very important feature in my opinion, as Kenton mentioned, we will be able to not only identify the data quality and correctness on the data, but take action as well and help isolate erroneous rows rather than failing the entire dataset.
Thanks for submitting this issue, @KentonParton, and the follow-up, @OmarSultan85 ! We will discuss internally, and I will get back to you in the next few days.
Hi @KentonParton and @OmarSultan85 - apologies for delays! A part of the V3 API was making sure that this functionality was available, but @jcampbell and I worked on this draft PR (#3346) to link up all the pieces and get it working. When you have a chance, we'd love your thoughts on the interface of it, concerns, etc.

Thanks!
Hi @talagluck ,

Thanks a lot for the update, I checked the PR but I am not entirely sure if I understood correctly. What I understand is that now the list of unexpected rows will be returned completely as part of the validation result when its of type COMPLETE.

If that is the case then I believe this would be of great use and would allows us to use great expectations in a pipeline that could split the incoming data into two, process correct rows and insert them and for the unexpected ones perform some kind of action  to clean the data, notify business users, etc.

Anyway I can try this new feature before it is released? I can't wait to integrate it into our product and pipelines :) 
Thanks so much for the feedback, @OmarSultan85 ! That is indeed the purpose of the feature, and I'm glad to hear that that will work for you!

We are still figuring out the correct user-interface for this, as it is possible that it will wind up returning far too much data, and so we would want to be explicit and careful so that we don't wind up including an entire table in validation, results for instance :) This is something that @KentonParton had called out, and is important to keep in mind.

What we're thinking right now is that we will add another `result_format` option (e.g. something like `INCLUDE_ALL_UNEXPECTED_ROWS`) that would return the entire contents of this. This would make sure that only users who definitely want this and know what they are doing would use this.

We are also still looking into specifying a list of columns that would be returned for each `unexpected_rows`. This is more along the lines of the original request for this issue. In this case, for instance, you would be able to say that where `column_a` has unexpected rows, return the contents of `column_b` and `column_c`. This seems achievable, but it goes pretty deep into the code (since it would affect all column_map expectations), and so we're working on it, but it will take some more time.
any updates on this?
Hi @talagluck I noticed just now that in #3764 that the GE team was planning on taking a look at this back in January.

Is progress being made against this issue on the GE jira board or has this been de-prioritized? If it has been de-prioritized (as tends to happen in our world) is this the kind of feature that the GE team would be able to offer guidance on?

Our dependence on pandas for the unexpected_index_list is fast becoming a constraint. We need this feature to be able to move to validating our data on a DB in order to allow us to continue to use GE as the size of our data scales.
Hi @abekfenn - thanks so much for the ping, and thanks for calling this out. This is something that I've been picking up as I have time (and I don't have a lot of time). I don't think this is a huge amount of work to implement, [especially in light of the work done in this PR for `unexpected_rows` ](https://github.com/great-expectations/great_expectations/pull/3346), it's just a matter of resources. If this is something on which you have the capacity to work, I'd be happy to offer guidance and support for design and code review. Let me know and we can set up a call.
Hi @talagluck thanks for getting back to me. Totally understandable, my work day sounds very similar 😄. I would be grateful if we could get on a call to discuss the scope of the work. I'll reach out over Slack. 
### Update
I've started working on this but run into an issue that I hope to figure out soon.

I do have some questions around the API that I would like GE/community feedback on.

With the introduction of `unexpected_rows` I'm torn between calling this new result format arg, `unexpected_index_columns` or `unexpected_row_columns`. The idea being that a user should provide to this arg a list of columns in the table that should be sub-setted from the unexpected rows.
This feature request was initially to support ID/PKs in the validation results, however I could see a use case whereby someone using the PandasExecutionEngine could want both unexpected_index_list and a subset of columns from the unexpected rows.

1. I see GE generally making use of binary args, but the arg `row_condition` led me to thinking this would be the cleanest naming convention in line with existing result format args, not to mention user interaction.
- However, I wanted to be sure that the GE team/community wouldn't prefer something like `include_unexpected_index_columns` AND `unexpected_index_columns`?
2. Similarly, I'm wondering if the returned list should be a new field in the validation results or if we should simply set `unexpected_index_list` when `unexpected_index_columns` is set?
- This would keep behavior consistent across SQLAlchemy and Pandas execution engines but the only concern is that people may want to use the same functionality for pandas - returning arbitrary columns rather than the pandas index list - and so that might create some confusion
- Alternatively, we could set `unexpected_index_list` only for SQLAlchemy but I think this only confuses matters, and it probably wouldn't do much to help keep behavior consistent across engines.

I'm personally leaning towards one arg for `unexpected_row_columns` and one new result field `unexpected_row_columns` or something alike but given that this deviates slightly from the original intention of this ticket, I wanted to open it for discussion.

Commit message: final update

Patch to review:
diff --git a/great_expectations/expectations/expectation.py b/great_expectations/expectations/expectation.py
index 00e241b6276c..572007059e75 100644
--- a/great_expectations/expectations/expectation.py
+++ b/great_expectations/expectations/expectation.py
@@ -73,6 +73,7 @@
 from great_expectations.execution_engine import (
     ExecutionEngine,
     PandasExecutionEngine,
+    SparkDFExecutionEngine,
     SqlAlchemyExecutionEngine,
 )
 from great_expectations.expectations.registry import (
@@ -2608,6 +2609,34 @@ def get_validation_dependencies(
                         metric_value_kwargs=metric_kwargs["metric_value_kwargs"],
                     ),
                 )
+        if isinstance(execution_engine, SparkDFExecutionEngine):
+            if "unexpected_index_column_names" in validation_dependencies.result_format:
+                metric_kwargs = get_metric_kwargs(
+                    f"{self.map_metric}.unexpected_index_list",
+                    configuration=configuration,
+                    runtime_configuration=runtime_configuration,
+                )
+                validation_dependencies.set_metric_configuration(
+                    metric_name=f"{self.map_metric}.unexpected_index_list",
+                    metric_configuration=MetricConfiguration(
+                        metric_name=f"{self.map_metric}.unexpected_index_list",
+                        metric_domain_kwargs=metric_kwargs["metric_domain_kwargs"],
+                        metric_value_kwargs=metric_kwargs["metric_value_kwargs"],
+                    ),
+                )
+                metric_kwargs = get_metric_kwargs(
+                    f"{self.map_metric}.unexpected_index_query",
+                    configuration=configuration,
+                    runtime_configuration=runtime_configuration,
+                )
+                validation_dependencies.set_metric_configuration(
+                    metric_name=f"{self.map_metric}.unexpected_index_query",
+                    metric_configuration=MetricConfiguration(
+                        metric_name=f"{self.map_metric}.unexpected_index_query",
+                        metric_domain_kwargs=metric_kwargs["metric_domain_kwargs"],
+                        metric_value_kwargs=metric_kwargs["metric_value_kwargs"],
+                    ),
+                )
         return validation_dependencies
 
     def _validate(
diff --git a/great_expectations/expectations/metrics/map_metric_provider.py b/great_expectations/expectations/metrics/map_metric_provider.py
index 418ba0bc52a5..fb96aa7e961e 100644
--- a/great_expectations/expectations/metrics/map_metric_provider.py
+++ b/great_expectations/expectations/metrics/map_metric_provider.py
@@ -498,7 +498,6 @@ def inner_func(
                     _compute_domain_kwargs=compute_domain_kwargs,
                     _accessor_domain_kwargs=accessor_domain_kwargs,
                 )
-
                 filter_column_isnull = kwargs.get(
                     "filter_column_isnull", getattr(cls, "filter_column_isnull", True)
                 )
@@ -2616,23 +2615,23 @@ def _spark_column_map_condition_value_counts(
     unexpected_condition, compute_domain_kwargs, accessor_domain_kwargs = metrics.get(
         "unexpected_condition"
     )
-    df = execution_engine.get_domain_records(domain_kwargs=compute_domain_kwargs)
-
     if "column" not in accessor_domain_kwargs:
         raise ValueError(
             """No "column" found in provided metric_domain_kwargs, but it is required for a column map metric
 (_spark_column_map_condition_value_counts).
 """
         )
-
     column_name: Union[str, quoted_name] = accessor_domain_kwargs["column"]
-
     column_name = get_dbms_compatible_column_names(
         column_names=column_name,
         batch_columns_list=metrics["table.columns"],
         execution_engine=execution_engine,
     )
 
+    df: "pyspark.sql.dataframe.DataFrame" = execution_engine.get_domain_records(
+        domain_kwargs=compute_domain_kwargs
+    )
+
     # withColumn is required to transform window functions returned by some metrics to boolean mask
     data = df.withColumn("__unexpected", unexpected_condition)
     filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
@@ -2677,6 +2676,120 @@ def _spark_map_condition_rows(
     return filtered.limit(result_format["partial_unexpected_count"]).collect()
 
 
+def _spark_map_condition_index(
+    cls,
+    execution_engine: SparkDFExecutionEngine,
+    metric_domain_kwargs: Dict,
+    metric_value_kwargs: Dict,
+    metrics: Dict[str, Any],
+    **kwargs,
+) -> Union[List[Dict[str, Any]], None]:
+    """
+    Returns indices of the metric values which do not meet an expected Expectation condition for instances
+    of ColumnMapExpectation.
+
+    Requires `unexpected_index_column_names` to be part of `result_format` dict to specify primary_key columns
+    to return.
+    """
+    (
+        unexpected_condition,
+        compute_domain_kwargs,
+        accessor_domain_kwargs,
+    ) = metrics.get("unexpected_condition", (None, None, None))
+
+    if unexpected_condition is None:
+        return None
+
+    if "column" not in accessor_domain_kwargs:
+        raise ValueError(
+            """No "column" found in provided metric_domain_kwargs, but it is required for a column map metric
+(_spark_column_map_condition_values).
+"""
+        )
+    domain_kwargs = dict(**compute_domain_kwargs, **accessor_domain_kwargs)
+    domain_column = domain_kwargs["column"]
+
+    df: "pyspark.sql.dataframe.DataFrame" = execution_engine.get_domain_records(
+        domain_kwargs=domain_kwargs
+    )
+
+    result_format = metric_value_kwargs["result_format"]
+    if not result_format.get("unexpected_index_column_names"):
+        raise ge_exceptions.MetricResolutionError(
+            "unexpected_indices cannot be returned without 'unexpected_index_column_names'. Please check your configuration."
+        )
+
+    # withColumn is required to transform window functions returned by some metrics to boolean mask
+    data = df.withColumn("__unexpected", unexpected_condition)
+    filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
+    unexpected_index_list: Optional[List[Dict[str, Any]]] = []
+
+    unexpected_index_column_names: List[str] = result_format[
+        "unexpected_index_column_names"
+    ]
+    columns_to_keep: List[str] = [column for column in unexpected_index_column_names]
+    columns_to_keep.append(domain_column)
+
+    # check that column name is in row
+    for col_name in columns_to_keep:
+        if col_name not in filtered.columns:
+            raise ge_exceptions.InvalidMetricAccessorDomainKwargsKeyError(
+                f"Error: The unexpected_index_column '{col_name}' does not exist in Spark DataFrame. Please check your configuration and try again."
+            )
+
+    for row in filtered.collect():
+        dict_to_add: dict = {}
+        for col_name in columns_to_keep:
+            dict_to_add[col_name] = row[col_name]
+        unexpected_index_list.append(dict_to_add)
+
+    if result_format["result_format"] == "COMPLETE":
+        return unexpected_index_list
+    return unexpected_index_list[: result_format["partial_unexpected_count"]]
+
+
+def _spark_map_condition_query(
+    cls,
+    execution_engine: SparkDFExecutionEngine,
+    metric_domain_kwargs: Dict,
+    metric_value_kwargs: Dict,
+    metrics: Dict[str, Any],
+    **kwargs,
+) -> Union[str, None]:
+    """
+    Returns query that will return all rows which do not meet an expected Expectation condition for instances
+    of ColumnMapExpectation.
+
+    Converts unexpected_condition into a string that can be rendered in DataDocs
+
+    Output will look like:
+
+        df.filter(F.expr( [unexpected_condition] ))
+
+    """
+    result_format: dict = metric_value_kwargs["result_format"]
+    # We will not return map_condition_query if return_unexpected_index_query = False
+    return_unexpected_index_query: bool = result_format.get(
+        "return_unexpected_index_query"
+    )
+    if return_unexpected_index_query is False:
+        return None
+
+    (
+        unexpected_condition,
+        _,
+        _,
+    ) = metrics.get("unexpected_condition", (None, None, None))
+
+    # unexpected_condition is an F.column object, meaning the str representation is wrapped in Column<> syntax.
+    # like Column<'[unexpected_expression]'>
+    unexpected_condition_as_string: str = str(unexpected_condition)
+    unexpected_condition_filtered: str = unexpected_condition_as_string.replace(
+        "Column<'(", ""
+    ).replace(")'>", "")
+    return f"df.filter(F.expr({unexpected_condition_filtered}))"
+
+
 def _spark_column_pair_map_condition_values(
     cls,
     execution_engine: SparkDFExecutionEngine,
@@ -3181,6 +3294,24 @@ def _register_metric_functions(cls):
                         metric_provider=_spark_map_condition_rows,
                         metric_fn_type=MetricFunctionTypes.VALUE,
                     )
+                    register_metric(
+                        metric_name=f"{metric_name}.unexpected_index_list",
+                        metric_domain_keys=metric_domain_keys,
+                        metric_value_keys=(*metric_value_keys, "result_format"),
+                        execution_engine=engine,
+                        metric_class=cls,
+                        metric_provider=_spark_map_condition_index,
+                        metric_fn_type=MetricFunctionTypes.VALUE,
+                    )
+                    register_metric(
+                        metric_name=f"{metric_name}.unexpected_index_query",
+                        metric_domain_keys=metric_domain_keys,
+                        metric_value_keys=(*metric_value_keys, "result_format"),
+                        execution_engine=engine,
+                        metric_class=cls,
+                        metric_provider=_spark_map_condition_query,
+                        metric_fn_type=MetricFunctionTypes.VALUE,
+                    )
                     if metric_fn_type == MetricPartialFunctionTypes.MAP_CONDITION_FN:
                         if domain_type == MetricDomainTypes.COLUMN:
                             register_metric(
diff --git a/tests/checkpoint/test_checkpoint_result_format.py b/tests/checkpoint/test_checkpoint_result_format.py
index 8e24eda4d740..fb72b8b6ff56 100644
--- a/tests/checkpoint/test_checkpoint_result_format.py
+++ b/tests/checkpoint/test_checkpoint_result_format.py
@@ -119,6 +119,26 @@ def batch_request_for_pandas_unexpected_rows_and_index(
 
 
 @pytest.fixture()
+def batch_request_for_spark_unexpected_rows_and_index(
+    spark_dataframe_for_unexpected_rows_with_index,
+) -> dict:
+    dataframe: "pyspark.sql.dataframe.DataFrame" = (
+        spark_dataframe_for_unexpected_rows_with_index
+    )
+    return {
+        "datasource_name": "spark_datasource",
+        "data_connector_name": "runtime_data_connector",
+        "data_asset_name": "IN_MEMORY_DATA_ASSET",
+        "runtime_parameters": {
+            "batch_data": dataframe,
+        },
+        "batch_identifiers": {
+            "id_key_0": 1234567890,
+        },
+    }
+
+
+@pytest.fixture
 def expected_unexpected_indices_output() -> List[Dict[str, Union[str, int]]]:
     return [
         {"animals": "giraffe", "pk_1": 3},
@@ -127,13 +147,16 @@ def expected_unexpected_indices_output() -> List[Dict[str, Union[str, int]]]:
     ]
 
 
-@pytest.fixture()
-def unexpected_index_query_output() -> str:
-    return (
-        "SELECT animals, pk_1 \n"
-        "FROM animal_names \n"
-        "WHERE animals IS NOT NULL AND (animals NOT IN ('cat', 'fish', 'dog'))"
-    )
+@pytest.fixture
+def expected_sql_query_output() -> str:
+    return "SELECT animals, pk_1 \n\
+FROM animal_names \n\
+WHERE animals IS NOT NULL AND (animals NOT IN ('cat', 'fish', 'dog'))"
+
+
+@pytest.fixture
+def expected_spark_query_output() -> str:
+    return "df.filter(F.expr((animals IS NOT NULL) AND (NOT (animals IN (cat, fish, dog)))))"
 
 
 def _add_expectations_and_checkpoint(
@@ -186,7 +209,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_complete_out
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
-    unexpected_index_query_output,
+    expected_sql_query_output,
 ):
     """
     What does this test?
@@ -231,7 +254,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_complete_out
     unexpected_index_query: str = evrs[0]["results"][0]["result"][
         "unexpected_index_query"
     ]
-    assert unexpected_index_query == unexpected_index_query_output
+    assert unexpected_index_query == expected_sql_query_output
 
 
 @pytest.mark.integration
@@ -240,7 +263,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_complete_out
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
-    unexpected_index_query_output,
+    expected_sql_query_output,
 ):
     """
     What does this test?
@@ -287,7 +310,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_complete_out
     unexpected_index_query: str = evrs[0]["results"][0]["result"][
         "unexpected_index_query"
     ]
-    assert unexpected_index_query == unexpected_index_query_output
+    assert unexpected_index_query == expected_sql_query_output
 
 
 @pytest.mark.integration
@@ -296,7 +319,6 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_complete_out
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
-    unexpected_index_query_output,
 ):
     """
     What does this test?
@@ -349,7 +371,7 @@ def test_sql_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expe
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
-    unexpected_index_query_output,
+    expected_sql_query_output,
 ):
     """
     What does this test?
@@ -390,7 +412,7 @@ def test_sql_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expe
     unexpected_index_query: str = evrs[0]["results"][0]["result"][
         "unexpected_index_query"
     ]
-    assert unexpected_index_query == unexpected_index_query_output
+    assert unexpected_index_query == expected_sql_query_output
 
 
 @pytest.mark.integration
@@ -398,7 +420,7 @@ def test_sql_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expe
     data_context_with_connection_to_animal_names_db,
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    unexpected_index_query_output,
+    expected_sql_query_output,
 ):
     """
     What does this test?
@@ -436,7 +458,7 @@ def test_sql_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expe
     unexpected_index_query: str = evrs[0]["results"][0]["result"][
         "unexpected_index_query"
     ]
-    assert unexpected_index_query == unexpected_index_query_output
+    assert unexpected_index_query == expected_sql_query_output
 
 
 @pytest.mark.integration
@@ -481,7 +503,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_two_expectation_complete_out
     expectation_config_expect_column_values_to_be_in_set,
     expectation_config_expect_column_values_to_not_be_in_set,
     expected_unexpected_indices_output,
-    unexpected_index_query_output,
+    expected_sql_query_output,
 ):
     """
     What does this test?
@@ -534,7 +556,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_two_expectation_complete_out
     unexpected_index_query: str = evrs[0]["results"][0]["result"][
         "unexpected_index_query"
     ]
-    assert unexpected_index_query == unexpected_index_query_output
+    assert unexpected_index_query == expected_sql_query_output
 
 
 @pytest.mark.integration
@@ -588,7 +610,6 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_basic_output
     data_context_with_connection_to_animal_names_db,
     reference_sql_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    unexpected_index_query_output,
 ):
     """
     What does this test?
@@ -888,19 +909,14 @@ def test_pandas_result_format_in_checkpoint_pk_defined_two_expectation_complete_
 
 
 @pytest.mark.integration
-def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_summary_output(
-    data_context_with_connection_to_animal_names_db,
-    reference_sql_checkpoint_config_for_unexpected_column_names,
+def test_pandas_result_format_in_checkpoint_pk_defined_one_expectation_summary_output(
+    in_memory_runtime_context,
+    batch_request_for_pandas_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
     unexpected_index_query_output,
 ):
-    """
-    What does this test?
-        - unexpected_index_column defined in Checkpoint only.
-        - SUMMARY output, which means we have `partial_unexpected_index_list` only
-        - 1 Expectations added to suite
-    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
             "result_format": "SUMMARY",
@@ -908,8 +924,8 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_summary_outp
         }
     }
     context: DataContext = _add_expectations_and_checkpoint(
-        data_context=data_context_with_connection_to_animal_names_db,
-        checkpoint_config=reference_sql_checkpoint_config_for_unexpected_column_names,
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
         expectations_list=[expectation_config_expect_column_values_to_be_in_set],
         dict_to_update_checkpoint=dict_to_update_checkpoint,
     )
@@ -917,6 +933,7 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_summary_outp
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
 
@@ -929,163 +946,92 @@ def test_sql_result_format_in_checkpoint_pk_defined_one_expectation_summary_outp
         "unexpected_index_list"
     )
     assert not first_result_full_list
-
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
-        "partial_unexpected_index_list"
-    ]
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
+        "result"
+    ].get("partial_unexpected_index_list")
     assert first_result_partial_list == expected_unexpected_indices_output
-    assert evrs[0]["results"][0]["result"].get("unexpected_index_query") is None
+    first_result_partial_list = evrs[0]["results"][0]["result"].get(
+        "partial_unexpected_index_list"
+    )
+    assert not first_result_partial_list
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_pk_defined_one_expectation_summary_output(
+def test_pandas_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output(
     in_memory_runtime_context,
     batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
 ):
-    dict_to_update_checkpoint: dict = {
-        "result_format": {
-            "result_format": "SUMMARY",
-            "unexpected_index_column_names": ["pk_1"],
-        }
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
         expectations_list=[expectation_config_expect_column_values_to_be_in_set],
-        dict_to_update_checkpoint=dict_to_update_checkpoint,
     )
-
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["pk_1"],
+    }
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
+        result_format=result_format,
         batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
-
-    index_column_names: List[str] = evrs[0]["results"][0]["result"][
-        "unexpected_index_column_names"
+    first_result_full_list = evrs[0]["results"][0]["result"]["unexpected_index_list"]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
     ]
-    assert index_column_names == ["pk_1"]
-
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_list"
-    )
-    assert not first_result_full_list
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
     assert first_result_partial_list == expected_unexpected_indices_output
+    assert evrs[0]["results"][0]["result"].get("unexpected_index_query") is None
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_one_index_column(
+def test_pandas_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_summary_output_limit_1(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    expected_unexpected_indices_output,
 ):
-    """
-    What does this test?
-        - DataFrame being passed into Checkpoint has named index 'pk_1', which correspond to unexpected_index_column_names
-        - MapMatric calculation happens the same as if `pk_1` was a non-index column
-    """
-    dict_to_update_checkpoint: dict = {
-        "result_format": {
-            "result_format": "COMPLETE",
-            "unexpected_index_column_names": ["pk_2"],
-        }
-    }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    # setting named index
-    updated_dataframe: pd.DataFrame = dataframe.set_index("pk_2")
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
         expectations_list=[expectation_config_expect_column_values_to_be_in_set],
-        dict_to_update_checkpoint=dict_to_update_checkpoint,
     )
-
+    result_format: dict = {
+        "result_format": "SUMMARY",
+        "partial_unexpected_count": 1,
+        "unexpected_index_column_names": ["pk_1"],
+    }
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request,
+        result_format=result_format,
+        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
-
-    index_column_names: List[str] = evrs[0]["results"][0]["result"][
-        "unexpected_index_column_names"
-    ]
-    assert index_column_names == ["pk_2"]
-
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_list"
-    )
-    assert first_result_full_list == [
-        {"animals": "giraffe", "pk_2": "three"},
-        {"animals": "lion", "pk_2": "four"},
-        {"animals": "zebra", "pk_2": "five"},
-    ]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
-    assert first_result_partial_list == [
-        {"animals": "giraffe", "pk_2": "three"},
-        {"animals": "lion", "pk_2": "four"},
-        {"animals": "zebra", "pk_2": "five"},
+    first_result_partial_list = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
     ]
+    assert first_result_partial_list == [{"animals": "giraffe", "pk_1": 3}]
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_one_index_column_wrong_column(
+def test_pandas_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output_incorrect_column(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    expected_unexpected_indices_output,
 ):
-    """
-    What does this test?
-        - DataFrame being passed into Checkpoint has named index 'pk_1', which doesn't match the unexpected_index_column_name
-    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
             "result_format": "COMPLETE",
-            "unexpected_index_column_names": ["pk_2"],  # I am not the index column
+            "unexpected_index_column_names": ["i_dont_exist"],
         }
     }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    # setting named index
-    updated_dataframe: pd.DataFrame = dataframe.set_index("pk_1")
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
@@ -1096,122 +1042,78 @@ def test_pandas_result_format_in_checkpoint_named_index_one_index_column_wrong_c
         result: CheckpointResult = context.run_checkpoint(
             checkpoint_name="my_checkpoint",
             expectation_suite_name="animal_names_exp",
-            batch_request=batch_request,
+            batch_request=batch_request_for_pandas_unexpected_rows_and_index,
             runtime_configuration={"catch_exceptions": False},
         )
     assert e.value.message == (
         "Exception occurred while running validation[0] of Checkpoint "
-        "'my_checkpoint': Error: The column pk_2 does not exist in the named indices. "
-        "Please check your configuration."
+        "'my_checkpoint': Error: The unexpected_index_column \"i_dont_exist\" does "
+        "not exist in Dataframe. Please check your configuration and try again.."
     )
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_two_index_column(
+def test_pandas_result_format_in_checkpoint_pk_defined_two_expectation_complete_output(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
+    expectation_config_expect_column_values_to_not_be_in_set,
     expected_unexpected_indices_output,
 ):
-    """
-    What does this test?
-        - DataFrame being passed into Checkpoint has two named indices, which correspond to unexpected_index_column_names
-        - MapMatric calculation happens the same as if `pk_1` and `pk_2` were non-index columns
-    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
             "result_format": "COMPLETE",
-            "unexpected_index_column_names": ["pk_1", "pk_2"],
+            "unexpected_index_column_names": ["pk_1"],
         }
     }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    # setting named index
-    updated_dataframe: pd.DataFrame = dataframe.set_index(["pk_1", "pk_2"])
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
-        expectations_list=[expectation_config_expect_column_values_to_be_in_set],
+        expectations_list=[
+            expectation_config_expect_column_values_to_be_in_set,
+            expectation_config_expect_column_values_to_not_be_in_set,
+        ],
         dict_to_update_checkpoint=dict_to_update_checkpoint,
     )
 
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request,
+        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
-
-    index_column_names: List[str] = evrs[0]["results"][0]["result"][
-        "unexpected_index_column_names"
+    # first and second expectations have same results. Although one is "expect_to_be"
+    # and the other is "expect_to_not_be", they have opposite value_sets
+    first_result_full_list = evrs[0]["results"][0]["result"]["unexpected_index_list"]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
     ]
-    assert index_column_names == ["pk_1", "pk_2"]
+    assert first_result_partial_list == expected_unexpected_indices_output
 
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_list"
-    )
-    assert first_result_full_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
-    ]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
-    assert first_result_partial_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
+    second_result_full_list = evrs[0]["results"][1]["result"]["unexpected_index_list"]
+    assert second_result_full_list == expected_unexpected_indices_output
+    second_result_partial_list = evrs[0]["results"][1]["result"][
+        "partial_unexpected_index_list"
     ]
+    assert second_result_partial_list == expected_unexpected_indices_output
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_two_index_column_not_set(
+def test_pandas_result_format_in_checkpoint_pk_defined_one_expectation_summary_output(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
 ):
-    """
-    What does this test?
-        - DataFrame being passed into Checkpoint has two named indices, which correspond to unexpected_index_column_names
-        - MapMatric calculation happens the same as if `pk_1` and `pk_2` were non-index columns
-    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
-            "result_format": "COMPLETE",
+            "result_format": "SUMMARY",
+            "unexpected_index_column_names": ["pk_1"],
         }
     }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    # setting named index
-    updated_dataframe: pd.DataFrame = dataframe.set_index(["pk_2"])
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
@@ -1222,58 +1124,32 @@ def test_pandas_result_format_in_checkpoint_named_index_two_index_column_not_set
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request,
+        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
-
-    index_column_names: List[str] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_column_names"
-    )
-    assert not index_column_names
-
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
+    first_result_full_list = evrs[0]["results"][0]["result"].get(
         "unexpected_index_list"
     )
-    assert first_result_full_list == [("three"), ("four"), ("five")]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
-    assert first_result_partial_list == [("three"), ("four"), ("five")]
+    assert not first_result_full_list
+    first_result_partial_list = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_two_index_column_not_set(
+def test_pandas_result_format_in_checkpoint_pk_defined_one_expectation_basic_output(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_pandas_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    expected_unexpected_indices_output,
 ):
-    """
-    What does this test?
-        - DataFrame being passed into Checkpoint has two named indices, which correspond to unexpected_index_column_names
-        - MapMatric calculation happens the same as if `pk_1` and `pk_2` were non-index columns
-    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
-            "result_format": "COMPLETE",
+            "result_format": "BASIC",
+            "unexpected_index_column_names": ["pk_1"],
         }
     }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    updated_dataframe: pd.DataFrame = dataframe.set_index(["pk_1", "pk_2"])
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
@@ -1284,125 +1160,357 @@ def test_pandas_result_format_in_checkpoint_named_index_two_index_column_not_set
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request,
+        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
-
-    index_column_names: List[str] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_column_names"
+    first_result_full_list = evrs[0]["results"][0]["result"].get(
+        "unexpected_index_list"
     )
-    assert not index_column_names
-
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
-        "unexpected_index_list"
+    assert not first_result_full_list
+    first_result_partial_list = evrs[0]["results"][0]["result"].get(
+        "partial_unexpected_index_list"
     )
-    assert first_result_full_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
-    ]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
-    assert first_result_partial_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
-    ]
+    assert not first_result_partial_list
 
 
+# spark
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_different_column_specified_in_result_format(
+def test_spark_result_format_in_checkpoint_pk_defined_one_expectation_complete_output(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_spark_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
     expected_unexpected_indices_output,
+    expected_spark_query_output,
 ):
     """
     What does this test?
-        - DataFrame being passed into Checkpoint has a different named index than then column passed in
-        - we also pass in `pk_1`  as unexpected_index_column_names
-        - but pk_2 is the actual index
+        - unexpected_index_column defined in Checkpoint only.
+        - COMPLETE output, which means we have `unexpected_index_list` and `partial_unexpected_index_list`
+        - 1 Expectations added to suite
     """
+
     dict_to_update_checkpoint: dict = {
         "result_format": {
             "result_format": "COMPLETE",
             "unexpected_index_column_names": ["pk_1"],
         }
     }
-    # build our own BatchRequest
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    # setting named index
-    updated_dataframe: pd.DataFrame = dataframe.set_index(["pk_2"])
+    context: DataContext = _add_expectations_and_checkpoint(
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
+        expectations_list=[expectation_config_expect_column_values_to_be_in_set],
+        dict_to_update_checkpoint=dict_to_update_checkpoint,
+    )
 
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
+    result: CheckpointResult = context.run_checkpoint(
+        checkpoint_name="my_checkpoint",
+        expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
+        runtime_configuration={"catch_exceptions": False},
+    )
+    evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
+    first_result_full_list = evrs[0]["results"][0]["result"]["unexpected_index_list"]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
+    unexpected_index_query: str = evrs[0]["results"][0]["result"][
+        "unexpected_index_query"
+    ]
+    assert unexpected_index_query == expected_spark_query_output
+
+
+@pytest.mark.integration
+def test_spark_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output(
+    in_memory_runtime_context,
+    batch_request_for_spark_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
+    expectation_config_expect_column_values_to_be_in_set,
+    expected_unexpected_indices_output,
+    expected_spark_query_output,
+):
+    """
+    What does this test?
+        - unexpected_index_column not defined in Checkpoint config, but passed in at run_checkpoint.
+        - COMPLETE output, which means we have `unexpected_index_list` and `partial_unexpected_index_list`
+        - 1 Expectations added to suite
+    """
+    # intentionally empty, since we are updating at run_checkpoint()
+    dict_to_update_checkpoint: dict = {}
+    context: DataContext = _add_expectations_and_checkpoint(
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
+        expectations_list=[expectation_config_expect_column_values_to_be_in_set],
+        dict_to_update_checkpoint=dict_to_update_checkpoint,
+    )
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["pk_1"],
+    }
+    result: CheckpointResult = context.run_checkpoint(
+        checkpoint_name="my_checkpoint",
+        expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
+        result_format=result_format,
+    )
+    evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
+
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
+        "unexpected_index_column_names"
+    ]
+    assert index_column_names == ["pk_1"]
+
+    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "unexpected_index_list"
+    ]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
+    unexpected_index_query: str = evrs[0]["results"][0]["result"][
+        "unexpected_index_query"
+    ]
+    assert unexpected_index_query == expected_spark_query_output
+
+
+@pytest.mark.integration
+def test_spark_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output_with_query(
+    in_memory_runtime_context,
+    batch_request_for_spark_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
+    expectation_config_expect_column_values_to_be_in_set,
+    expected_unexpected_indices_output,
+    expected_spark_query_output,
+):
+    """
+    What does this test?
+        - unexpected_index_column not defined in Checkpoint config, but passed in at run_checkpoint.
+        - COMPLETE output, which means we have `unexpected_index_list` and `partial_unexpected_index_list`
+        - 1 Expectations added to suite
+        - return_unexpected_index_query set to True
+    """
+    # intentionally empty, since we are updating at run_checkpoint()
+    dict_to_update_checkpoint: dict = {}
+    context: DataContext = _add_expectations_and_checkpoint(
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
+        expectations_list=[expectation_config_expect_column_values_to_be_in_set],
+        dict_to_update_checkpoint=dict_to_update_checkpoint,
+    )
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["pk_1"],
+        "return_unexpected_index_query": True,
     }
+    result: CheckpointResult = context.run_checkpoint(
+        checkpoint_name="my_checkpoint",
+        expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
+        result_format=result_format,
+    )
+    evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
+
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
+        "unexpected_index_column_names"
+    ]
+    assert index_column_names == ["pk_1"]
+
+    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "unexpected_index_list"
+    ]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
+
+    unexpected_index_query = evrs[0]["results"][0]["result"].get(
+        "unexpected_index_query"
+    )
+    assert unexpected_index_query == expected_spark_query_output
+
+
+@pytest.mark.integration
+def test_spark_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output_no_query(
+    in_memory_runtime_context,
+    batch_request_for_spark_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
+    expectation_config_expect_column_values_to_be_in_set,
+    expected_unexpected_indices_output,
+):
+    """
+    What does this test?
+        - unexpected_index_column not defined in Checkpoint config, but passed in at run_checkpoint.
+        - COMPLETE output, which means we have `unexpected_index_list` and `partial_unexpected_index_list`
+        - 1 Expectations added to suite
+        - return_unexpected_index_query set to False
+    """
+    # intentionally empty, since we are updating at run_checkpoint()
+    dict_to_update_checkpoint: dict = {}
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
         expectations_list=[expectation_config_expect_column_values_to_be_in_set],
         dict_to_update_checkpoint=dict_to_update_checkpoint,
     )
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["pk_1"],
+        "return_unexpected_index_query": False,
+    }
+    result: CheckpointResult = context.run_checkpoint(
+        checkpoint_name="my_checkpoint",
+        expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
+        result_format=result_format,
+    )
+    evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
 
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
+        "unexpected_index_column_names"
+    ]
+    assert index_column_names == ["pk_1"]
+
+    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "unexpected_index_list"
+    ]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
+    assert evrs[0]["results"][0]["result"].get("unexpected_index_query") is None
+
+
+@pytest.mark.integration
+def test_spark_result_format_not_in_checkpoint_passed_into_run_checkpoint_one_expectation_complete_output_incorrect_column(
+    in_memory_runtime_context,
+    batch_request_for_spark_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
+    expectation_config_expect_column_values_to_be_in_set,
+):
+    """
+    What does this test?
+       - unexpected_index_column not defined in Checkpoint config, but passed in at run_checkpoint.
+       - unexpected_index_column is passed in an incorrect column
+    """
+    context: DataContext = _add_expectations_and_checkpoint(
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
+        expectations_list=[expectation_config_expect_column_values_to_be_in_set],
+    )
+
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["i_dont_exist"],
+    }
     with pytest.raises(CheckpointError) as e:
         result: CheckpointResult = context.run_checkpoint(
             checkpoint_name="my_checkpoint",
             expectation_suite_name="animal_names_exp",
-            batch_request=batch_request,
+            batch_request=batch_request_for_spark_unexpected_rows_and_index,
+            result_format=result_format,
             runtime_configuration={"catch_exceptions": False},
         )
+
     assert e.value.message == (
         "Exception occurred while running validation[0] of Checkpoint "
-        "'my_checkpoint': Error: The column pk_1 does not exist in the named indices. "
-        "Please check your configuration."
+        "'my_checkpoint': Error: The unexpected_index_column 'i_dont_exist' does not "
+        "exist in Spark DataFrame. Please check your configuration and try again.."
     )
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_named_index_two_index_column_set(
+def test_spark_result_format_in_checkpoint_pk_defined_two_expectation_complete_output(
     in_memory_runtime_context,
-    pandas_animals_dataframe_for_unexpected_rows_and_index,
+    batch_request_for_spark_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
+    expectation_config_expect_column_values_to_not_be_in_set,
     expected_unexpected_indices_output,
+    expected_spark_query_output,
 ):
     """
     What does this test?
-        - DataFrame being passed into Checkpoint has two named indices, which correspond to unexpected_index_column_names
-        - MapMetric calculation happens the same as if `pk_1` and `pk_2` were non-index columns
-        - we also pass in `pk_1` and `pk_2`  as unexpected_index_column_names
+        - unexpected_index_column not defined in Checkpoint config, but passed in at run_checkpoint.
+        - COMPLETE output, which means we have `unexpected_index_list` and `partial_unexpected_index_list`
+        - 2 Expectations added to suite
+    """
+    context: DataContext = _add_expectations_and_checkpoint(
+        data_context=in_memory_runtime_context,
+        checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
+        expectations_list=[
+            expectation_config_expect_column_values_to_be_in_set,
+            expectation_config_expect_column_values_to_not_be_in_set,
+        ],
+    )
+    result_format: dict = {
+        "result_format": "COMPLETE",
+        "unexpected_index_column_names": ["pk_1"],
+    }
+
+    result: CheckpointResult = context.run_checkpoint(
+        checkpoint_name="my_checkpoint",
+        expectation_suite_name="animal_names_exp",
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
+        result_format=result_format,
+    )
+    evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
+
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
+        "unexpected_index_column_names"
+    ]
+    assert index_column_names == ["pk_1"]
+
+    # first and second expectations have same results. Although one is "expect_to_be"
+    # and the other is "expect_to_not_be", they have opposite value_sets
+    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "unexpected_index_list"
+    ]
+    assert first_result_full_list == expected_unexpected_indices_output
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert first_result_partial_list == expected_unexpected_indices_output
 
+    second_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][1]["result"][
+        "unexpected_index_list"
+    ]
+    assert second_result_full_list == expected_unexpected_indices_output
+
+    second_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][1]["result"][
+        "partial_unexpected_index_list"
+    ]
+    assert second_result_partial_list == expected_unexpected_indices_output
+    unexpected_index_query: str = evrs[0]["results"][0]["result"][
+        "unexpected_index_query"
+    ]
+    assert unexpected_index_query == expected_spark_query_output
+
+
+@pytest.mark.integration
+def test_spark_result_format_in_checkpoint_pk_defined_one_expectation_summary_output(
+    in_memory_runtime_context,
+    batch_request_for_spark_unexpected_rows_and_index,
+    reference_checkpoint_config_for_unexpected_column_names,
+    expectation_config_expect_column_values_to_be_in_set,
+    expected_unexpected_indices_output,
+):
+    """
+    What does this test?
+        - unexpected_index_column defined in Checkpoint only.
+        - SUMMARY output, which means we have `partial_unexpected_index_list` only
+        - 1 Expectations added to suite
     """
     dict_to_update_checkpoint: dict = {
         "result_format": {
-            "result_format": "COMPLETE",
-            "unexpected_index_column_names": ["pk_1", "pk_2"],
+            "result_format": "SUMMARY",
+            "unexpected_index_column_names": ["pk_1"],
         }
     }
-    dataframe: pd.DataFrame = pandas_animals_dataframe_for_unexpected_rows_and_index
-    updated_dataframe: pd.DataFrame = dataframe.set_index(["pk_1", "pk_2"])
-
-    batch_request: dict = {
-        "datasource_name": "pandas_datasource",
-        "data_connector_name": "runtime_data_connector",
-        "data_asset_name": "IN_MEMORY_DATA_ASSET",
-        "runtime_parameters": {
-            "batch_data": updated_dataframe,
-        },
-        "batch_identifiers": {
-            "id_key_0": 1234567890,
-        },
-    }
     context: DataContext = _add_expectations_and_checkpoint(
         data_context=in_memory_runtime_context,
         checkpoint_config=reference_checkpoint_config_for_unexpected_column_names,
@@ -1413,45 +1521,43 @@ def test_pandas_result_format_in_checkpoint_named_index_two_index_column_set(
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request,
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
 
-    index_column_names: List[str] = evrs[0]["results"][0]["result"].get(
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
         "unexpected_index_column_names"
-    )
-    assert index_column_names == ["pk_1", "pk_2"]
+    ]
+    assert index_column_names == ["pk_1"]
 
     first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
         "unexpected_index_list"
     )
-    assert first_result_full_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
-    ]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0][
-        "result"
-    ].get("partial_unexpected_index_list")
-    assert first_result_partial_list == [
-        {"animals": "giraffe", "pk_1": 3, "pk_2": "three"},
-        {"animals": "lion", "pk_1": 4, "pk_2": "four"},
-        {"animals": "zebra", "pk_1": 5, "pk_2": "five"},
+    assert not first_result_full_list
+    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+        "partial_unexpected_index_list"
     ]
+    assert first_result_partial_list == expected_unexpected_indices_output
+    assert evrs[0]["results"][0]["result"].get("unexpected_index_query") is None
 
 
 @pytest.mark.integration
-def test_pandas_result_format_in_checkpoint_one_expectation_complete_output(
+def test_spark_result_format_in_checkpoint_pk_defined_one_expectation_basic_output(
     in_memory_runtime_context,
-    batch_request_for_pandas_unexpected_rows_and_index,
+    batch_request_for_spark_unexpected_rows_and_index,
     reference_checkpoint_config_for_unexpected_column_names,
     expectation_config_expect_column_values_to_be_in_set,
-    expected_unexpected_indices_output,
 ):
-    """ """
+    """
+    What does this test?
+        - unexpected_index_column defined in Checkpoint only.
+        - BASIC output, which means we have no unexpected_index_list output
+        - 1 Expectations added to suite
+    """
     dict_to_update_checkpoint: dict = {
         "result_format": {
-            "result_format": "COMPLETE",
+            "result_format": "BASIC",
+            "unexpected_index_column_names": ["pk_1"],
         }
     }
     context: DataContext = _add_expectations_and_checkpoint(
@@ -1464,15 +1570,21 @@ def test_pandas_result_format_in_checkpoint_one_expectation_complete_output(
     result: CheckpointResult = context.run_checkpoint(
         checkpoint_name="my_checkpoint",
         expectation_suite_name="animal_names_exp",
-        batch_request=batch_request_for_pandas_unexpected_rows_and_index,
+        batch_request=batch_request_for_spark_unexpected_rows_and_index,
     )
     evrs: List[ExpectationSuiteValidationResult] = result.list_validation_results()
 
-    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
-        "unexpected_index_list"
+    index_column_names: List[str] = evrs[0]["results"][0]["result"][
+        "unexpected_index_column_names"
     ]
-    assert first_result_full_list == [3, 4, 5]
-    first_result_partial_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"][
+    assert index_column_names == ["pk_1"]
+
+    first_result_full_list: List[Dict[str, Any]] = evrs[0]["results"][0]["result"].get(
+        "unexpected_index_list"
+    )
+    assert not first_result_full_list
+    first_result_partial_list = evrs[0]["results"][0]["result"].get(
         "partial_unexpected_index_list"
-    ]
-    assert first_result_partial_list == [3, 4, 5]
+    )
+    assert not first_result_partial_list
+    assert evrs[0]["results"][0]["result"].get("unexpected_index_query") is None
diff --git a/tests/conftest.py b/tests/conftest.py
index d20064ece6a9..1f598d578de1 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -7239,3 +7239,27 @@ def pandas_animals_dataframe_for_unexpected_rows_and_index():
             ],
         }
     )
+
+
+@pytest.fixture
+def spark_dataframe_for_unexpected_rows_with_index(
+    spark_session,
+) -> "pyspark.sql.dataframe.DataFrame":  # noqa: F821
+    df: pd.DataFrame = pd.DataFrame(
+        {
+            "pk_1": [0, 1, 2, 3, 4, 5],
+            "pk_2": ["zero", "one", "two", "three", "four", "five"],
+            "animals": [
+                "cat",
+                "fish",
+                "dog",
+                "giraffe",
+                "lion",
+                "zebra",
+            ],
+        }
+    )
+    test_df: "pyspark.sql.dataframe.DataFrame" = spark_session.createDataFrame(
+        data=df,
+    )  # noqa: F821
+    return test_df
diff --git a/tests/expectations/metrics/test_map_metric.py b/tests/expectations/metrics/test_map_metric.py
index b7d31bf05aaf..4262045d8a61 100644
--- a/tests/expectations/metrics/test_map_metric.py
+++ b/tests/expectations/metrics/test_map_metric.py
@@ -14,7 +14,6 @@
 from great_expectations.data_context.util import file_relative_path
 from great_expectations.datasource import Datasource
 from great_expectations.datasource.data_connector import ConfiguredAssetSqlDataConnector
-from great_expectations.exceptions import InvalidExpectationConfigurationError
 from great_expectations.execution_engine import (
     PandasExecutionEngine,
     SparkDFExecutionEngine,
@@ -37,30 +36,6 @@
 from great_expectations.validator.validator import Validator
 
 
-@pytest.fixture
-def spark_dataframe_for_unexpected_rows_with_index(
-    spark_session,
-) -> "pyspark.sql.dataframe.DataFrame":  # noqa: F821
-    df: pandas.DataFrame = pd.DataFrame(
-        {
-            "pk_1": [0, 1, 2, 3, 4, 5],
-            "pk_2": ["zero", "one", "two", "three", "four", "five"],
-            "animals": [
-                "cat",
-                "fish",
-                "dog",
-                "giraffe",
-                "lion",
-                "zebra",
-            ],
-        }
-    )
-    test_df: "pyspark.sql.dataframe.DataFrame" = spark_session.createDataFrame(
-        data=df
-    )  # noqa: F821
-    return test_df
-
-
 @pytest.fixture
 def sqlite_table_for_unexpected_rows_with_index(
     test_backends,
@@ -825,112 +800,6 @@ def test_spark_single_column_summary_result_format(
     }
 
 
-# Spark
-def test_spark_single_column_complete_result_format(
-    in_memory_runtime_context,
-    spark_dataframe_for_unexpected_rows_with_index,
-):
-    expectation_configuration = ExpectationConfiguration(
-        expectation_type="expect_column_values_to_be_in_set",
-        kwargs={
-            "column": "animals",
-            "value_set": ["cat", "fish", "dog"],
-            "result_format": {
-                "result_format": "COMPLETE",
-            },
-        },
-    )
-    expectation = ExpectColumnValuesToBeInSet(expectation_configuration)
-    batch_definition = BatchDefinition(
-        datasource_name="spark_datasource",
-        data_connector_name="runtime_data_connector",
-        data_asset_name="my_asset",
-        batch_identifiers=IDDict({}),
-        batch_spec_passthrough=None,
-    )
-    batch = Batch(
-        data=spark_dataframe_for_unexpected_rows_with_index,
-        batch_definition=batch_definition,
-    )
-    engine = SparkDFExecutionEngine()
-    validator = Validator(
-        execution_engine=engine,
-        data_context=in_memory_runtime_context,
-        batches=[
-            batch,
-        ],
-    )
-    result = expectation.validate(validator)
-    assert convert_to_json_serializable(result.result) == {
-        "element_count": 6,
-        "missing_count": 0,
-        "missing_percent": 0.0,
-        "partial_unexpected_counts": [
-            {"count": 1, "value": "giraffe"},
-            {"count": 1, "value": "lion"},
-            {"count": 1, "value": "zebra"},
-        ],
-        "partial_unexpected_list": ["giraffe", "lion", "zebra"],
-        "unexpected_count": 3,
-        "unexpected_list": ["giraffe", "lion", "zebra"],
-        "unexpected_percent": 50.0,
-        "unexpected_percent_nonmissing": 50.0,
-        "unexpected_percent_total": 50.0,
-    }
-
-
-def test_spark_single_column_summary_result_format(
-    in_memory_runtime_context,
-    spark_dataframe_for_unexpected_rows_with_index,
-):
-    expectation_configuration = ExpectationConfiguration(
-        expectation_type="expect_column_values_to_be_in_set",
-        kwargs={
-            "column": "animals",
-            "value_set": ["cat", "fish", "dog"],
-            "result_format": {
-                "result_format": "SUMMARY",
-            },
-        },
-    )
-    expectation = ExpectColumnValuesToBeInSet(expectation_configuration)
-    batch_definition = BatchDefinition(
-        datasource_name="spark_datasource",
-        data_connector_name="runtime_data_connector",
-        data_asset_name="my_asset",
-        batch_identifiers=IDDict({}),
-        batch_spec_passthrough=None,
-    )
-    batch = Batch(
-        data=spark_dataframe_for_unexpected_rows_with_index,
-        batch_definition=batch_definition,
-    )
-    engine = SparkDFExecutionEngine()
-    validator = Validator(
-        execution_engine=engine,
-        data_context=in_memory_runtime_context,
-        batches=[
-            batch,
-        ],
-    )
-    result = expectation.validate(validator)
-    assert convert_to_json_serializable(result.result) == {
-        "element_count": 6,
-        "missing_count": 0,
-        "missing_percent": 0.0,
-        "partial_unexpected_counts": [
-            {"count": 1, "value": "giraffe"},
-            {"count": 1, "value": "lion"},
-            {"count": 1, "value": "zebra"},
-        ],
-        "partial_unexpected_list": ["giraffe", "lion", "zebra"],
-        "unexpected_count": 3,
-        "unexpected_percent": 50.0,
-        "unexpected_percent_nonmissing": 50.0,
-        "unexpected_percent_total": 50.0,
-    }
-
-
 def test_sqlite_single_column_complete_result_format(
     sa,
     in_memory_runtime_context,


Code review comments:
diff_hunk: @@ -2677,6 +2676,120 @@ def _spark_map_condition_rows(
     return filtered.limit(result_format["partial_unexpected_count"]).collect()
 
 
+def _spark_map_condition_index(
+    cls,
+    execution_engine: SparkDFExecutionEngine,
+    metric_domain_kwargs: Dict,
+    metric_value_kwargs: Dict,
+    metrics: Dict[str, Any],
+    **kwargs,
+) -> Union[List[Dict[str, Any]], None]:
+    """
+    Returns indices of the metric values which do not meet an expected Expectation condition for instances
+    of ColumnMapExpectation.
+
+    Requires `unexpected_index_column_names` to be part of `result_format` dict to specify primary_key columns
+    to return.
+    """
+    (
+        unexpected_condition,
+        compute_domain_kwargs,
+        accessor_domain_kwargs,
+    ) = metrics.get("unexpected_condition", (None, None, None))
+
+    if unexpected_condition is None:
+        return None
+
+    if "column" not in accessor_domain_kwargs:
+        raise ValueError(
+            """No "column" found in provided metric_domain_kwargs, but it is required for a column map metric
+(_spark_column_map_condition_values).
+"""
+        )
+    domain_kwargs = dict(**compute_domain_kwargs, **accessor_domain_kwargs)
+    domain_column = domain_kwargs["column"]
+
+    df: "pyspark.sql.dataframe.DataFrame" = execution_engine.get_domain_records(
+        domain_kwargs=domain_kwargs
+    )
+
+    result_format = metric_value_kwargs["result_format"]
+    if not result_format.get("unexpected_index_column_names"):
+        raise ge_exceptions.MetricResolutionError(
+            "unexpected_indices cannot be returned without 'unexpected_index_column_names'. Please check your configuration."
+        )
+
+    # withColumn is required to transform window functions returned by some metrics to boolean mask
+    data = df.withColumn("__unexpected", unexpected_condition)
+    filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
+    unexpected_index_list: Optional[List[Dict[str, Any]]] = []
+
+    unexpected_index_column_names: List[str] = result_format[
+        "unexpected_index_column_names"
+    ]
+    columns_to_keep: List[str] = [column for column in unexpected_index_column_names]
+    columns_to_keep.append(domain_column)
+
+    # check that column name is in row
+    for col_name in columns_to_keep:
+        if col_name not in filtered.columns:
+            raise ge_exceptions.InvalidMetricAccessorDomainKwargsKeyError(
+                f"Error: The unexpected_index_column '{col_name}' does not exist in Spark DataFrame. Please check your configuration and try again."
+            )
+
+    for row in filtered.collect():
+        dict_to_add: dict = {}
+        for col_name in columns_to_keep:
+            dict_to_add[col_name] = row[col_name]
+        unexpected_index_list.append(dict_to_add)
+
+    if result_format["result_format"] == "COMPLETE":
+        return unexpected_index_list
+    return unexpected_index_list[: result_format["partial_unexpected_count"]]
+
path: great_expectations/expectations/metrics/map_metric_provider.py
text: @user1:
Do you think we can do the filtering before the collect statement for the case where we are not returning the COMPLETE results? Otherwise it looks like here we are retrieving all of the unexpected values to the machine running GX and then filtering them if COMPLETE is not specified. I'm just concerned that this may cause undesired extra computation / transfer (especially for large datasets).

@user1:
It would be good to double check for this in the SQL case as well.

@author:
great idea. I'll look into making the change both here and at the SQL stage

@author:
This is already done for SQL

diff_hunk: @@ -2608,6 +2609,34 @@ def get_validation_dependencies(
                         metric_value_kwargs=metric_kwargs["metric_value_kwargs"],
                     ),
                 )
+        if isinstance(execution_engine, SparkDFExecutionEngine):
path: great_expectations/expectations/expectation.py
text: @user1:
Are these sections the same? If so maybe you could change the if to check either SQL or Spark e.g. `if isinstance(execution_engine, (SqlAlchemyExecutionEngine, SparkDFExecutionEngine)):`

@author:
great catch. Thank you :)