class_docstr: These test cases exercise the interface to the proto plan
  generation but do not call Spark.
class_docstr_tok: ["These","test","cases","exercise","the","interface","to","the","proto","plan","generation","but","do","not","call","Spark","."]
code: def test_window_functions(self):
      self.assertEqual(CW.unboundedPreceding, SW.unboundedPreceding)
  
      self.assertEqual(CW.unboundedFollowing, SW.unboundedFollowing)
  
      self.assertEqual(CW.currentRow, SW.currentRow)
  
      query = """
              SELECT * FROM VALUES
              (0, float("NAN"), NULL), (1, NULL, 2.0), (1, 2.1, 3.5), (0, 0.5, 1.0),
              (0, 1.5, 1.1), (1, 2.2, -1.0), (1, 0.1, -0.1), (0, 0.0, 5.0)
              AS tab(a, b, c)
              """
      # +---+----+----+
      # |  a|   b|   c|
      # +---+----+----+
      # |  0| NaN|NULL|
      # |  1|NULL| 2.0|
      # |  1| 2.1| 3.5|
      # |  0| 0.5| 1.0|
      # |  0| 1.5| 1.1|
      # |  1| 2.2|-1.0|
      # |  1| 0.1|-0.1|
      # |  0| 0.0| 5.0|
      # +---+----+----+
  
      cdf = self.connect.sql(query)
      sdf = self.spark.sql(query)
  
      # test window functions
      for ccol, scol in [
          (CF.row_number(), SF.row_number()),
          (CF.rank(), SF.rank()),
          (CF.dense_rank(), SF.dense_rank()),
          (CF.percent_rank(), SF.percent_rank()),
          (CF.cume_dist(), SF.cume_dist()),
          (CF.lag("c", 1), SF.lag("c", 1)),
          (CF.lag("c", 1, -1.0), SF.lag("c", 1, -1.0)),
          (CF.lag(cdf.c, -1), SF.lag(sdf.c, -1)),
          (
              CF.lag(cdf.c, -1, float("nan")),
              SF.lag(sdf.c, -1, float("nan")),
          ),
          (CF.lead("c", 1), SF.lead("c", 1)),
          (CF.lead("c", 1, -1.0), SF.lead("c", 1, -1.0)),
          (CF.lead(cdf.c, -1), SF.lead(sdf.c, -1)),
          (
              CF.lead(cdf.c, -1, float("nan")),
              SF.lead(sdf.c, -1, float("nan")),
          ),
          (CF.nth_value("c", 1), SF.nth_value("c", 1)),
          (CF.nth_value(cdf.c, 2), SF.nth_value(sdf.c, 2)),
          (CF.nth_value(cdf.c, 2, True), SF.nth_value(sdf.c, 2, True)),
          (
              CF.nth_value(cdf.c, 2, False),
              SF.nth_value(sdf.c, 2, False),
          ),
          (CF.ntile(1), SF.ntile(1)),
          (CF.ntile(2), SF.ntile(2)),
          (CF.ntile(4), SF.ntile(4)),
      ]:
          for cwin, swin in [
              (CW.orderBy("b"), SW.orderBy("b")),
              (
                  CW.partitionBy("a").orderBy("b"),
                  SW.partitionBy("a").orderBy("b"),
              ),
              (
                  CW.partitionBy("a").orderBy(CF.col("b").desc()),
                  SW.partitionBy("a").orderBy(SF.col("b").desc()),
              ),
              (
                  CW.partitionBy("a", cdf.c).orderBy("b"),
                  SW.partitionBy("a", sdf.c).orderBy("b"),
              ),
              (
                  CW.partitionBy("a").orderBy("b", cdf.c),
                  SW.partitionBy("a").orderBy("b", sdf.c),
              ),
              (
                  CW.partitionBy("a").orderBy("b", cdf.c.desc()),
                  SW.partitionBy("a").orderBy("b", sdf.c.desc()),
              ),
          ]:
              self.assert_eq(
                  cdf.select(ccol.over(cwin)).toPandas(),
                  sdf.select(scol.over(swin)).toPandas(),
              )
  
      # test aggregation functions
      for ccol, scol in [
          (CF.count("c"), SF.count("c")),
          (CF.sum("c"), SF.sum("c")),
          (CF.max(cdf.c), SF.max(sdf.c)),
          (CF.min(cdf.c), SF.min(sdf.c)),
      ]:
          for cwin, swin in [
              (CW.orderBy("b"), SW.orderBy("b")),
              (
                  CW.orderBy("b").rowsBetween(
                      CW.currentRow, CW.currentRow
                  ),
                  SW.orderBy("b").rowsBetween(
                      SW.currentRow, SW.currentRow
                  ),
              ),
              (
                  CW.orderBy(cdf.b.desc()).rowsBetween(
                      CW.currentRow - 1, CW.currentRow + 2
                  ),
                  SW.orderBy(sdf.b.desc()).rowsBetween(
                      SW.currentRow - 1, SW.currentRow + 2
                  ),
              ),
              (
                  CW.orderBy("b").rowsBetween(
                      CW.unboundedPreceding, CW.currentRow
                  ),
                  SW.orderBy("b").rowsBetween(
                      SW.unboundedPreceding, SW.currentRow
                  ),
              ),
              (
                  CW.orderBy(cdf.b.desc()).rowsBetween(
                      CW.currentRow, CW.unboundedFollowing
                  ),
                  SW.orderBy(sdf.b.desc()).rowsBetween(
                      SW.currentRow, SW.unboundedFollowing
                  ),
              ),
              (
                  CW.orderBy("b").rangeBetween(
                      CW.currentRow, CW.currentRow
                  ),
                  SW.orderBy("b").rangeBetween(
                      SW.currentRow, SW.currentRow
                  ),
              ),
              (
                  CW.orderBy("b").rangeBetween(
                      CW.currentRow - 1, CW.currentRow + 2
                  ),
                  SW.orderBy("b").rangeBetween(
                      SW.currentRow - 1, SW.currentRow + 2
                  ),
              ),
              (
                  CW.orderBy("b").rangeBetween(
                      CW.unboundedPreceding, CW.currentRow
                  ),
                  SW.orderBy("b").rangeBetween(
                      SW.unboundedPreceding, SW.currentRow
                  ),
              ),
              (
                  CW.orderBy("b").rangeBetween(
                      CW.currentRow, CW.unboundedFollowing
                  ),
                  SW.orderBy("b").rangeBetween(
                      SW.currentRow, SW.unboundedFollowing
                  ),
              ),
              (
                  CW.partitionBy("a").orderBy("b"),
                  SW.partitionBy("a").orderBy("b"),
              ),
              (
                  CW.partitionBy(cdf.a)
                  .orderBy(CF.asc_nulls_last("b"))
                  .rowsBetween(CW.currentRow, CW.currentRow),
                  SW.partitionBy(sdf.a)
                  .orderBy(SF.asc_nulls_last("b"))
                  .rowsBetween(SW.currentRow, SW.currentRow),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy(cdf.b.desc())
                  .rowsBetween(CW.currentRow - 1, CW.currentRow + 2),
                  SW.partitionBy("a")
                  .orderBy(sdf.b.desc())
                  .rowsBetween(SW.currentRow - 1, SW.currentRow + 2),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy("b")
                  .rowsBetween(CW.unboundedPreceding, CW.currentRow),
                  SW.partitionBy("a")
                  .orderBy("b")
                  .rowsBetween(SW.unboundedPreceding, SW.currentRow),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy("b")
                  .rowsBetween(CW.currentRow, CW.unboundedFollowing),
                  SW.partitionBy("a")
                  .orderBy("b")
                  .rowsBetween(SW.currentRow, SW.unboundedFollowing),
              ),
              (
                  CW.partitionBy(cdf.a)
                  .orderBy(cdf.b.desc(), "c")
                  .rangeBetween(CW.currentRow, CW.currentRow),
                  SW.partitionBy(sdf.a)
                  .orderBy(sdf.b.desc(), "c")
                  .rangeBetween(SW.currentRow, SW.currentRow),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy("b")
                  .rangeBetween(CW.currentRow - 1, CW.currentRow + 2),
                  SW.partitionBy("a")
                  .orderBy("b")
                  .rangeBetween(SW.currentRow - 1, SW.currentRow + 2),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy(CF.desc_nulls_last("b"))
                  .rangeBetween(CW.unboundedPreceding, CW.currentRow),
                  SW.partitionBy("a")
                  .orderBy(SF.desc_nulls_last("b"))
                  .rangeBetween(SW.unboundedPreceding, SW.currentRow),
              ),
              (
                  CW.partitionBy("a")
                  .orderBy("b")
                  .rangeBetween(CW.currentRow, CW.unboundedFollowing),
                  SW.partitionBy("a")
                  .orderBy("b")
                  .rangeBetween(SW.currentRow, SW.unboundedFollowing),
              ),
          ]:
              self.assert_eq(
                  cdf.select(ccol.over(cwin)).toPandas(),
                  sdf.select(scol.over(swin)).toPandas(),
              )
  
      # check error
      with self.assertRaises(PySparkValueError) as pe:
          cdf.select(
              CF.sum("a").over(
                  CW.orderBy("b").rowsBetween(0, (1 << 33))
              )
          ).show()
  
      self.check_error(
          exception=pe.exception,
          error_class="VALUE_NOT_BETWEEN",
          message_parameters={
              "arg_name": "end",
              "min": "-2147483648",
              "max": "2147483647",
          },
      )
  
      with self.assertRaises(PySparkTypeError) as pe:
          cdf.select(CF.rank().over(cdf.a))
  
      self.check_error(
          exception=pe.exception,
          error_class="NOT_WINDOWSPEC",
          message_parameters={
              "arg_name": "window",
              "arg_type": "Column",
          },
      )
  
      # invalid window function
      with self.assertRaises(AnalysisException):
          cdf.select(cdf.b.over(CW.orderBy("b"))).show()
  
      # invalid window frame
      # following functions require Windowframe(RowFrame, UnboundedPreceding, CurrentRow)
      for ccol in [
          CF.row_number(),
          CF.rank(),
          CF.dense_rank(),
          CF.percent_rank(),
          CF.lag("c", 1),
          CF.lead("c", 1),
          CF.ntile(1),
      ]:
          with self.assertRaises(AnalysisException):
              cdf.select(
                  ccol.over(
                      CW.orderBy("b").rowsBetween(
                          CW.currentRow, CW.currentRow + 123
                      )
                  )
              ).show()
  
          with self.assertRaises(AnalysisException):
              cdf.select(
                  ccol.over(
                      CW.orderBy("b").rangeBetween(
                          CW.currentRow, CW.currentRow + 123
                      )
                  )
              ).show()
  
          with self.assertRaises(AnalysisException):
              cdf.select(
                  ccol.over(
                      CW.orderBy("b").rangeBetween(
                          CW.unboundedPreceding, CW.currentRow
                      )
                  )
              ).show()
  
      # Function 'cume_dist' requires Windowframe(RangeFrame, UnboundedPreceding, CurrentRow)
      ccol = CF.cume_dist()
      with self.assertRaises(AnalysisException):
          cdf.select(
              ccol.over(
                  CW.orderBy("b").rangeBetween(
                      CW.currentRow, CW.currentRow + 123
                  )
              )
          ).show()
  
      with self.assertRaises(AnalysisException):
          cdf.select(
              ccol.over(
                  CW.orderBy("b").rowsBetween(
                      CW.currentRow, CW.currentRow + 123
                  )
              )
          ).show()
  
      with self.assertRaises(AnalysisException):
          cdf.select(
              ccol.over(
                  CW.orderBy("b").rowsBetween(
                      CW.unboundedPreceding, CW.currentRow
                  )
              )
          ).show()
  
imports: import os
  import unittest
  from inspect import getmembers, isfunction
  from pyspark.errors import PySparkTypeError, PySparkValueError
  from pyspark.sql import SparkSession
  from pyspark.sql.types import StringType, StructType, StructField, ArrayType, IntegerType
  from pyspark.testing import assertDataFrameEqual
  from pyspark.testing.pandasutils import PandasOnSparkTestUtils
  from pyspark.testing.connectutils import ReusedConnectTestCase, should_test_connect
  from pyspark.testing.sqlutils import SQLTestUtils
  from pyspark.errors.exceptions.connect import AnalysisException, SparkConnectException
label_desc: Use image node_id 12 for calling the SparkConnectFunctionTests obj's underlying member method code with example usage: obj.test_window_functions() without return types