code: def test_window_aggregate_with_pandas_udaf(self):
      import datetime
      from pyflink.table.window import Tumble
  
      t = self.t_env.from_elements(
          [
              (1, 2, 3, datetime.datetime(2018, 3, 11, 3, 10, 0, 0)),
              (3, 2, 4, datetime.datetime(2018, 3, 11, 3, 10, 0, 0)),
              (2, 1, 2, datetime.datetime(2018, 3, 11, 3, 10, 0, 0)),
              (1, 3, 1, datetime.datetime(2018, 3, 11, 3, 40, 0, 0)),
              (1, 8, 5, datetime.datetime(2018, 3, 11, 4, 20, 0, 0)),
              (2, 3, 6, datetime.datetime(2018, 3, 11, 3, 30, 0, 0)),
          ],
          DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.TINYINT()),
                  DataTypes.FIELD("b", DataTypes.SMALLINT()),
                  DataTypes.FIELD("c", DataTypes.INT()),
                  DataTypes.FIELD("rowtime", DataTypes.TIMESTAMP(3)),
              ]
          ),
      )
  
      sink_table_ddl = """
              CREATE TABLE Results_test_window_aggregate_with_pandas_udaf(
                  a TIMESTAMP(3),
                  b FLOAT,
                  c INT
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      print(t.get_schema())
      pandas_udaf = udaf(
          lambda pd: (pd.b.mean(), pd.b.max()),
          result_type=DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.FLOAT()),
                  DataTypes.FIELD("b", DataTypes.INT()),
              ]
          ),
          func_type="pandas",
      )
      tumble_window = (
          Tumble.over(expr.lit(1).hours)
          .on(expr.col("rowtime"))
          .alias("w")
      )
      t.select(t.b, t.rowtime).window(tumble_window).group_by(
          expr.col("w")
      ).aggregate(pandas_udaf.alias("d", "e")).select(
          expr.col("w").rowtime, expr.col("d"), expr.col("e")
      ).execute_insert(
          "Results_test_window_aggregate_with_pandas_udaf"
      ).wait()
  
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[2018-03-11 03:59:59.999, 2.2, 3]",
              "+I[2018-03-11 04:59:59.999, 8.0, 8]",
          ],
      )
  
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 3 for calling the BatchRowBasedOperationITTests obj's underlying member method code with example usage: obj.test_window_aggregate_with_pandas_udaf() without return types