code: def test_tumble_group_window_aggregate_function(self):
      from pyflink.table.window import Tumble
  
      # create source file path
      data = [
          "1,2,3,2018-03-11 03:10:00",
          "3,2,4,2018-03-11 03:10:00",
          "2,1,2,2018-03-11 03:10:00",
          "1,3,1,2018-03-11 03:40:00",
          "1,8,5,2018-03-11 04:20:00",
          "2,3,6,2018-03-11 03:30:00",
      ]
      source_path = (
          self.tempdir
          + "/test_tumble_group_window_aggregate_function.csv"
      )
      with open(source_path, "w") as fd:
          for ele in data:
              fd.write(ele + "\n")
  
      self.t_env.get_config().set(
          "pipeline.time-characteristic", "EventTime"
      )
      source_table = generate_random_table_name()
      source_table_ddl = f"""
              create table {source_table}(
                  a TINYINT,
                  b SMALLINT,
                  c INT,
                  rowtime TIMESTAMP(3),
                  WATERMARK FOR rowtime AS rowtime - INTERVAL '60' MINUTE
              ) with(
                  'connector.type' = 'filesystem',
                  'format.type' = 'csv',
                  'connector.path' = '{source_path}',
                  'format.ignore-first-line' = 'false',
                  'format.field-delimiter' = ','
              )
          """
      self.t_env.execute_sql(source_table_ddl)
      t = self.t_env.from_path(source_table)
      sink_table = generate_random_table_name()
      sink_table_ddl = f"""
              CREATE TABLE {sink_table}(
                  a TIMESTAMP(3),
                  b TIMESTAMP(3),
                  c FLOAT
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      tumble_window = (
          Tumble.over(lit(1).hours).on(col("rowtime")).alias("w")
      )
      t.window(tumble_window).group_by(col("w")).select(
          col("w").start, col("w").end, mean_udaf(t.b)
      ).execute_insert(sink_table).wait()
  
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[2018-03-11T03:00, 2018-03-11T04:00, 2.2]",
              "+I[2018-03-11T04:00, 2018-03-11T05:00, 8.0]",
          ],
      )
  
imports: import uuid
  from pyflink.table.expressions import col, call, lit, row_interval
  from pyflink.table.types import DataTypes
  from pyflink.table.udf import udaf, udf, AggregateFunction
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkBatchTableTestCase, PyFlinkStreamTableTestCase
label_desc: Use image node_id 6 for calling the BatchPandasUDAFITTests obj's underlying member method code with example usage: obj.test_tumble_group_window_aggregate_function() without return types