code: def test_complex_expressions(self):
      df = self.data
      plus_one = self.python_plus_one
      plus_two = self.pandas_scalar_plus_two
      sum_udf = self.pandas_agg_sum_udf
  
      # Test complex expressions with sql expression, python UDF and
      # group aggregate pandas UDF
      result1 = (
          df.withColumn("v1", plus_one(df.v))
          .withColumn("v2", df.v + 2)
          .groupby(df.id, df.v % 2)
          .agg(
              sum_udf(col("v")),
              sum_udf(col("v1") + 3),
              sum_udf(col("v2")) + 5,
              plus_one(sum_udf(col("v1"))),
              sum_udf(plus_one(col("v2"))),
          )
          .sort(["id", "(v % 2)"])
          .toPandas()
          .sort_values(by=["id", "(v % 2)"])
      )
  
      expected1 = (
          df.withColumn("v1", df.v + 1)
          .withColumn("v2", df.v + 2)
          .groupby(df.id, df.v % 2)
          .agg(
              sum(col("v")),
              sum(col("v1") + 3),
              sum(col("v2")) + 5,
              plus_one(sum(col("v1"))),
              sum(plus_one(col("v2"))),
          )
          .sort(["id", "(v % 2)"])
          .toPandas()
          .sort_values(by=["id", "(v % 2)"])
      )
  
      # Test complex expressions with sql expression, scala pandas UDF and
      # group aggregate pandas UDF
      result2 = (
          df.withColumn("v1", plus_one(df.v))
          .withColumn("v2", df.v + 2)
          .groupby(df.id, df.v % 2)
          .agg(
              sum_udf(col("v")),
              sum_udf(col("v1") + 3),
              sum_udf(col("v2")) + 5,
              plus_two(sum_udf(col("v1"))),
              sum_udf(plus_two(col("v2"))),
          )
          .sort(["id", "(v % 2)"])
          .toPandas()
          .sort_values(by=["id", "(v % 2)"])
      )
  
      expected2 = (
          df.withColumn("v1", df.v + 1)
          .withColumn("v2", df.v + 2)
          .groupby(df.id, df.v % 2)
          .agg(
              sum(col("v")),
              sum(col("v1") + 3),
              sum(col("v2")) + 5,
              plus_two(sum(col("v1"))),
              sum(plus_two(col("v2"))),
          )
          .sort(["id", "(v % 2)"])
          .toPandas()
          .sort_values(by=["id", "(v % 2)"])
      )
  
      # Test sequential groupby aggregate
      result3 = (
          df.groupby("id")
          .agg(sum_udf(df.v).alias("v"))
          .groupby("id")
          .agg(sum_udf(col("v")))
          .sort("id")
          .toPandas()
      )
  
      expected3 = (
          df.groupby("id")
          .agg(sum(df.v).alias("v"))
          .groupby("id")
          .agg(sum(col("v")))
          .sort("id")
          .toPandas()
      )
  
      assert_frame_equal(expected1, result1)
      assert_frame_equal(expected2, result2)
      assert_frame_equal(expected3, result3)
  
imports: import unittest
  from typing import cast
  from pyspark.rdd import PythonEvalType
  from pyspark.sql import Row
  from pyspark.sql.functions import array, explode, col, lit, mean, sum, udf, pandas_udf, PandasUDFType
  from pyspark.sql.types import ArrayType, YearMonthIntervalType
  from pyspark.errors import AnalysisException, PySparkNotImplementedError, PythonException
  from pyspark.testing.sqlutils import ReusedSQLTestCase, have_pandas, have_pyarrow, pandas_requirement_message, pyarrow_requirement_message
  from pyspark.testing.utils import QuietTest, assertDataFrameEqual
label_desc: Use image node_id 16 for calling the GroupedAggPandasUDFTestsMixin obj's underlying member method code with example usage: obj.test_complex_expressions() without return types