[SPARK-40339][SPARK-40342][SPARK-40345][SPARK-40348][PS] Implement quantile in Rolling/RollingGroupby/Expanding/ExpandingGroupby

python/pyspark/pandas/groupby.py

@@ -607,7 +607,7 @@ def quantile(self, q: float = 0.5, accuracy: int = 10000) -> FrameLike:
         -------
         `quantile` in pandas-on-Spark are using distributed percentile approximation
         algorithm unlike pandas, the result might different with pandas, also
-        `interpolation` parameters are not supported yet.
+        `interpolation` parameter is not supported yet.
 
         See Also
         --------

python/pyspark/pandas/missing/window.py

@@ -82,7 +82,6 @@ class MissingPandasLikeExpanding:
     corr = _unsupported_function_expanding("corr")
     cov = _unsupported_function_expanding("cov")
     median = _unsupported_function_expanding("median")
-    quantile = _unsupported_function_expanding("quantile")
     validate = _unsupported_function_expanding("validate")
 
     exclusions = _unsupported_property_expanding("exclusions")
@@ -101,7 +100,6 @@ class MissingPandasLikeRolling:
     corr = _unsupported_function_rolling("corr")
     cov = _unsupported_function_rolling("cov")
     median = _unsupported_function_rolling("median")
-    quantile = _unsupported_function_rolling("quantile")
     validate = _unsupported_function_rolling("validate")
 
     exclusions = _unsupported_property_rolling("exclusions")
@@ -120,7 +118,6 @@ class MissingPandasLikeExpandingGroupby:
     corr = _unsupported_function_expanding("corr")
     cov = _unsupported_function_expanding("cov")
     median = _unsupported_function_expanding("median")
-    quantile = _unsupported_function_expanding("quantile")
     validate = _unsupported_function_expanding("validate")
 
     exclusions = _unsupported_property_expanding("exclusions")
@@ -139,7 +136,6 @@ class MissingPandasLikeRollingGroupby:
     corr = _unsupported_function_rolling("corr")
     cov = _unsupported_function_rolling("cov")
     median = _unsupported_function_rolling("median")
-    quantile = _unsupported_function_rolling("quantile")
     validate = _unsupported_function_rolling("validate")
 
     exclusions = _unsupported_property_rolling("exclusions")

python/pyspark/pandas/tests/test_expanding.py

@@ -82,6 +82,9 @@ def test_expanding_max(self):
     def test_expanding_mean(self):
         self._test_expanding_func("mean")
 
+    def test_expanding_quantile(self):
+        self._test_expanding_func(lambda x: x.quantile(0.5), lambda x: x.quantile(0.5, "lower"))
+
     def test_expanding_sum(self):
         self._test_expanding_func("sum")
 
@@ -212,6 +215,11 @@ def test_groupby_expanding_max(self):
     def test_groupby_expanding_mean(self):
         self._test_groupby_expanding_func("mean")
 
+    def test_groupby_expanding_quantile(self):
+        self._test_groupby_expanding_func(
+            lambda x: x.quantile(0.5), lambda x: x.quantile(0.5, "lower")
+        )
+
     def test_groupby_expanding_sum(self):
         self._test_groupby_expanding_func("sum")
 

python/pyspark/pandas/tests/test_rolling.py

@@ -79,6 +79,9 @@ def test_rolling_max(self):
     def test_rolling_mean(self):
         self._test_rolling_func("mean")
 
+    def test_rolling_quantile(self):
+        self._test_rolling_func(lambda x: x.quantile(0.5), lambda x: x.quantile(0.5, "lower"))
+
     def test_rolling_sum(self):
         self._test_rolling_func("sum")
 
@@ -212,6 +215,11 @@ def test_groupby_rolling_max(self):
     def test_groupby_rolling_mean(self):
         self._test_groupby_rolling_func("mean")
 
+    def test_groupby_rolling_quantile(self):
+        self._test_groupby_rolling_func(
+            lambda x: x.quantile(0.5), lambda x: x.quantile(0.5, "lower")
+        )
+
     def test_groupby_rolling_sum(self):
         self._test_groupby_rolling_func("sum")