code: def test_slide_group_window_aggregate_function(self):
      from pyflink.table.window import Slide
  
      # 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_slide_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 TINYINT,
                  b TIMESTAMP(3),
                  c TIMESTAMP(3),
                  d FLOAT,
                  e INT
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      slide_window = (
          Slide.over(lit(1).hours)
          .every(lit(30).minutes)
          .on(col("rowtime"))
          .alias("w")
      )
      t.window(slide_window).group_by(t.a, col("w")).select(
          t.a,
          col("w").start,
          col("w").end,
          mean_udaf(t.b),
          call("max_add", t.b, t.c, 1),
      ).execute_insert(sink_table).wait()
      actual = source_sink_utils.results()
  
      self.assert_equals(
          actual,
          [
              "+I[1, 2018-03-11T02:30, 2018-03-11T03:30, 2.0, 6]",
              "+I[1, 2018-03-11T03:00, 2018-03-11T04:00, 2.5, 7]",
              "+I[1, 2018-03-11T03:30, 2018-03-11T04:30, 5.5, 14]",
              "+I[1, 2018-03-11T04:00, 2018-03-11T05:00, 8.0, 14]",
              "+I[2, 2018-03-11T02:30, 2018-03-11T03:30, 1.0, 4]",
              "+I[2, 2018-03-11T03:00, 2018-03-11T04:00, 2.0, 10]",
              "+I[2, 2018-03-11T03:30, 2018-03-11T04:30, 3.0, 10]",
              "+I[3, 2018-03-11T03:00, 2018-03-11T04:00, 2.0, 7]",
              "+I[3, 2018-03-11T02:30, 2018-03-11T03:30, 2.0, 7]",
          ],
      )
  
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 7 for calling the BatchPandasUDAFITTests obj's underlying member method code with example usage: obj.test_slide_group_window_aggregate_function() without return types