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_collection_functions(self):
      query = """
              SELECT * FROM VALUES
              (ARRAY('a', 'ab'), ARRAY(1, 2, 3), ARRAY(1, NULL, 3), 1, 2, 'a'),
              (ARRAY('x', NULL), NULL, ARRAY(1, 3), 3, 4, 'x'),
              (NULL, ARRAY(-1, -2, -3), Array(), 5, 6, NULL)
              AS tab(a, b, c, d, e, f)
              """
      # +---------+------------+------------+---+---+----+
      # |        a|           b|           c|  d|  e|   f|
      # +---------+------------+------------+---+---+----+
      # |  [a, ab]|   [1, 2, 3]|[1, null, 3]|  1|  2|   a|
      # |[x, null]|        NULL|      [1, 3]|  3|  4|   x|
      # |     NULL|[-1, -2, -3]|          []|  5|  6|NULL|
      # +---------+------------+------------+---+---+----+
  
      cdf = self.connect.sql(query)
      sdf = self.spark.sql(query)
  
      for cfunc, sfunc in [
          (CF.array_distinct, SF.array_distinct),
          (CF.array_compact, SF.array_compact),
          (CF.array_max, SF.array_max),
          (CF.array_min, SF.array_min),
          (CF.reverse, SF.reverse),
          (CF.size, SF.size),
      ]:
          self.assert_eq(
              cdf.select(cfunc("a"), cfunc(cdf.b)).toPandas(),
              sdf.select(sfunc("a"), sfunc(sdf.b)).toPandas(),
          )
  
      for cfunc, sfunc in [
          (CF.array_except, SF.array_except),
          (CF.array_intersect, SF.array_intersect),
          (CF.array_union, SF.array_union),
          (CF.arrays_overlap, SF.arrays_overlap),
      ]:
          self.assert_eq(
              cdf.select(cfunc("b", cdf.c)).toPandas(),
              sdf.select(sfunc("b", sdf.c)).toPandas(),
          )
  
      for cfunc, sfunc in [
          (CF.array_position, SF.array_position),
          (CF.array_remove, SF.array_remove),
      ]:
          self.assert_eq(
              cdf.select(cfunc(cdf.a, "ab")).toPandas(),
              sdf.select(sfunc(sdf.a, "ab")).toPandas(),
          )
  
      # test array
      self.assert_eq(
          cdf.select(CF.array(cdf.d, "e")).toPandas(),
          sdf.select(SF.array(sdf.d, "e")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array(cdf.d, "e", CF.lit(99))).toPandas(),
          sdf.select(SF.array(sdf.d, "e", SF.lit(99))).toPandas(),
      )
  
      # test array_contains
      self.assert_eq(
          cdf.select(CF.array_contains(cdf.a, "ab")).toPandas(),
          sdf.select(SF.array_contains(sdf.a, "ab")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_contains(cdf.a, cdf.f)).toPandas(),
          sdf.select(SF.array_contains(sdf.a, sdf.f)).toPandas(),
      )
  
      # test array_append
      self.assert_eq(
          cdf.select(CF.array_append(cdf.a, "xyz")).toPandas(),
          sdf.select(SF.array_append(sdf.a, "xyz")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_append(cdf.a, CF.lit("ab"))).toPandas(),
          sdf.select(SF.array_append(sdf.a, SF.lit("ab"))).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_append(cdf.a, cdf.f)).toPandas(),
          sdf.select(SF.array_append(sdf.a, sdf.f)).toPandas(),
      )
  
      # test array_prepend
      self.assert_eq(
          cdf.select(CF.array_prepend(cdf.a, "xyz")).toPandas(),
          sdf.select(SF.array_prepend(sdf.a, "xyz")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_prepend(cdf.a, CF.lit("ab"))).toPandas(),
          sdf.select(SF.array_prepend(sdf.a, SF.lit("ab"))).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_prepend(cdf.a, cdf.f)).toPandas(),
          sdf.select(SF.array_prepend(sdf.a, sdf.f)).toPandas(),
      )
  
      # test array_insert
      self.assert_eq(
          cdf.select(CF.array_insert(cdf.a, -5, "ab")).toPandas(),
          sdf.select(SF.array_insert(sdf.a, -5, "ab")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_insert(cdf.a, 3, cdf.f)).toPandas(),
          sdf.select(SF.array_insert(sdf.a, 3, sdf.f)).toPandas(),
      )
  
      # test array_join
      self.assert_eq(
          cdf.select(
              CF.array_join(cdf.a, ","),
              CF.array_join("b", ":"),
              CF.array_join("c", "~"),
          ).toPandas(),
          sdf.select(
              SF.array_join(sdf.a, ","),
              SF.array_join("b", ":"),
              SF.array_join("c", "~"),
          ).toPandas(),
      )
      self.assert_eq(
          cdf.select(
              CF.array_join(cdf.a, ",", "_null_"),
              CF.array_join("b", ":", ".null."),
              CF.array_join("c", "~", "NULL"),
          ).toPandas(),
          sdf.select(
              SF.array_join(sdf.a, ",", "_null_"),
              SF.array_join("b", ":", ".null."),
              SF.array_join("c", "~", "NULL"),
          ).toPandas(),
      )
  
      # test array_repeat
      self.assert_eq(
          cdf.select(CF.array_repeat(cdf.f, "d")).toPandas(),
          sdf.select(SF.array_repeat(sdf.f, "d")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_repeat("f", cdf.d)).toPandas(),
          sdf.select(SF.array_repeat("f", sdf.d)).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.array_repeat("f", 3)).toPandas(),
          sdf.select(SF.array_repeat("f", 3)).toPandas(),
      )
  
      # test arrays_zip
      # TODO: Make toPandas support complex nested types like Array<Struct>
      # DataFrame.iloc[:, 0] (column name="arrays_zip(b, c)") values are different (66.66667 %)
      # [index]: [0, 1, 2]
      # [left]:  [[{'b': 1, 'c': 1.0}, {'b': 2, 'c': None}, {'b': 3, 'c': 3.0}], None,
      #           [{'b': -1, 'c': None}, {'b': -2, 'c': None}, {'b': -3, 'c': None}]]
      # [right]: [[(1, 1), (2, None), (3, 3)], None, [(-1, None), (-2, None), (-3, None)]]
      self.compare_by_show(
          cdf.select(CF.arrays_zip(cdf.b, "c")),
          sdf.select(SF.arrays_zip(sdf.b, "c")),
      )
  
      # test concat
      self.assert_eq(
          cdf.select(CF.concat("d", cdf.e, CF.lit(-1))).toPandas(),
          sdf.select(SF.concat("d", sdf.e, SF.lit(-1))).toPandas(),
      )
  
      # test create_map
      self.compare_by_show(
          cdf.select(CF.create_map(cdf.d, cdf.e)),
          sdf.select(SF.create_map(sdf.d, sdf.e)),
      )
      self.compare_by_show(
          cdf.select(CF.create_map(cdf.d, "e", "e", CF.lit(1))),
          sdf.select(SF.create_map(sdf.d, "e", "e", SF.lit(1))),
      )
  
      # test element_at
      self.assert_eq(
          cdf.select(CF.element_at("a", 1)).toPandas(),
          sdf.select(SF.element_at("a", 1)).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.element_at(cdf.a, 1)).toPandas(),
          sdf.select(SF.element_at(sdf.a, 1)).toPandas(),
      )
  
      # test get
      self.assert_eq(
          cdf.select(CF.get("a", 1)).toPandas(),
          sdf.select(SF.get("a", 1)).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.get(cdf.a, 1)).toPandas(),
          sdf.select(SF.get(sdf.a, 1)).toPandas(),
      )
  
      # test shuffle
      # Can not compare the values due to the random permutation
      self.assertEqual(
          cdf.select(CF.shuffle(cdf.a), CF.shuffle("b")).count(),
          sdf.select(SF.shuffle(sdf.a), SF.shuffle("b")).count(),
      )
  
      # test slice
      self.assert_eq(
          cdf.select(
              CF.slice(cdf.a, 1, 2), CF.slice("c", 2, 3)
          ).toPandas(),
          sdf.select(
              SF.slice(sdf.a, 1, 2), SF.slice("c", 2, 3)
          ).toPandas(),
      )
  
      with self.assertRaises(PySparkTypeError) as pe:
          CF.slice(cdf.a, 1.0, 2)
  
      self.check_error(
          exception=pe.exception,
          error_class="NOT_COLUMN_OR_INT_OR_STR",
          message_parameters={"arg_name": "start", "arg_type": "float"},
      )
  
      with self.assertRaises(PySparkTypeError) as pe:
          CF.slice(cdf.a, 1, 2.0)
  
      self.check_error(
          exception=pe.exception,
          error_class="NOT_COLUMN_OR_INT_OR_STR",
          message_parameters={
              "arg_name": "length",
              "arg_type": "float",
          },
      )
  
      # test sort_array
      self.assert_eq(
          cdf.select(
              CF.sort_array(cdf.a, True), CF.sort_array("c", False)
          ).toPandas(),
          sdf.select(
              SF.sort_array(sdf.a, True), SF.sort_array("c", False)
          ).toPandas(),
      )
  
      # test struct
      self.compare_by_show(
          cdf.select(CF.struct(cdf.a, "d", "e", cdf.f)),
          sdf.select(SF.struct(sdf.a, "d", "e", sdf.f)),
      )
  
      # test sequence
      self.assert_eq(
          cdf.select(CF.sequence(CF.lit(1), CF.lit(5))).toPandas(),
          sdf.select(SF.sequence(SF.lit(1), SF.lit(5))).toPandas(),
      )
      self.assert_eq(
          cdf.select(
              CF.sequence(CF.lit(1), CF.lit(5), CF.lit(1))
          ).toPandas(),
          sdf.select(
              SF.sequence(SF.lit(1), SF.lit(5), SF.lit(1))
          ).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.sequence(cdf.d, "e")).toPandas(),
          sdf.select(SF.sequence(sdf.d, "e")).toPandas(),
      )
      self.assert_eq(
          cdf.select(CF.sequence(cdf.d, "e", CF.lit(1))).toPandas(),
          sdf.select(SF.sequence(sdf.d, "e", SF.lit(1))).toPandas(),
      )
  
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 14 for calling the SparkConnectFunctionTests obj's underlying member method code with example usage: obj.test_collection_functions() without return types