tests.py 文件源码

def test_unbounded_frames(self):
        from unittest.mock import patch
        from pyspark.sql import functions as F
        from pyspark.sql import window
        import importlib

        df = self.spark.range(0, 3)

        def rows_frame_match():
            return "ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING" in df.select(
                F.count("*").over(window.Window.rowsBetween(-sys.maxsize, sys.maxsize))
            ).columns[0]

        def range_frame_match():
            return "RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING" in df.select(
                F.count("*").over(window.Window.rangeBetween(-sys.maxsize, sys.maxsize))
            ).columns[0]

        with patch("sys.maxsize", 2 ** 31 - 1):
            importlib.reload(window)
            self.assertTrue(rows_frame_match())
            self.assertTrue(range_frame_match())

        with patch("sys.maxsize", 2 ** 63 - 1):
            importlib.reload(window)
            self.assertTrue(rows_frame_match())
            self.assertTrue(range_frame_match())

        with patch("sys.maxsize", 2 ** 127 - 1):
            importlib.reload(window)
            self.assertTrue(rows_frame_match())
            self.assertTrue(range_frame_match())

        importlib.reload(window)