code: def test_aggregate_with_pandas_udaf(self):
      t = self.t_env.from_elements(
          [(1, 2, 3), (2, 1, 3), (1, 5, 4), (1, 8, 6), (2, 3, 4)],
          DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.TINYINT()),
                  DataTypes.FIELD("b", DataTypes.SMALLINT()),
                  DataTypes.FIELD("c", DataTypes.INT()),
              ]
          ),
      )
  
      sink_table_ddl = """
              CREATE TABLE Results_test_aggregate_with_pandas_udaf(
                  a TINYINT,
                  b FLOAT,
                  c INT
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      pandas_udaf = udaf(
          lambda pd: (pd.b.mean(), pd.a.max()),
          result_type=DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.FLOAT()),
                  DataTypes.FIELD("b", DataTypes.INT()),
              ]
          ),
          func_type="pandas",
      )
      t.select(t.a, t.b).group_by(t.a).aggregate(pandas_udaf).select(
          expr.col("*")
      ).execute_insert("Results_test_aggregate_with_pandas_udaf").wait()
      actual = source_sink_utils.results()
      self.assert_equals(actual, ["+I[1, 5.0, 1]", "+I[2, 2.0, 2]"])
  
imports: import unittest
  from pandas.testing import assert_frame_equal
  from pyflink.common import Row
  from pyflink.table import expressions, ListView
  from pyflink.table.types import DataTypes
  from pyflink.table.udf import udf, udtf, udaf, AggregateFunction, TableAggregateFunction, udtaf
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkBatchTableTestCase, PyFlinkStreamTableTestCase
label_desc: Use image node_id 1 for calling the BatchRowBasedOperationITTests obj's underlying member method code with example usage: obj.test_aggregate_with_pandas_udaf() without return types