[MAINTENANCE] Performance improvement refactor for Spark unexpected values

docs_rtd/changelog.rst

@@ -9,6 +9,7 @@ develop
 -----------------
 * [FEATURE] Configurable multi-threaded checkpoint speedup (#3362)
 * [DOCS] "Deploying Great Expectations in a hosted environment without file system or CLI" (#3361)
+* [MAINTENANCE] Spark performance improvement for metrics that return unexpected values (#3368)
 
 0.13.33
 -----------------

great_expectations/expectations/metrics/map_metric_provider.py

@@ -2373,17 +2373,10 @@ def _spark_column_map_condition_values(
             message=f'Error: The column "{column_name}" in BatchData does not exist.'
         )
 
-    data = (
-        df.withColumn("__row_number", F.row_number().over(Window.orderBy(F.lit(1))))
-        .withColumn("__unexpected", unexpected_condition)
-        .orderBy(F.col("__row_number"))
-    )
-
-    filtered = (
-        data.filter(F.col("__unexpected") == True)
-        .drop(F.col("__unexpected"))
-        .drop(F.col("__row_number"))
-    )
+    # withColumn is required to transform windowFunctions returned by some metrics to boolean mask
+    # e.g. increasing, decreasing, unique
+    data = df.withColumn("__unexpected", unexpected_condition)
+    filtered = data.filter(F.col("__unexpected") == True).drop(F.col("__unexpected"))
 
     result_format = metric_value_kwargs["result_format"]
     if result_format["result_format"] == "COMPLETE":