code: def test_sliding_group_window_over_time(self):
      # create source file path
      import tempfile
      import os
  
      tmp_dir = tempfile.gettempdir()
      data = [
          "1,1,2,2018-03-11 03:10:00",
          "3,3,2,2018-03-11 03:10:00",
          "2,2,1,2018-03-11 03:10:00",
          "1,1,3,2018-03-11 03:40:00",
          "1,1,8,2018-03-11 04:20:00",
          "2,2,3,2018-03-11 03:30:00",
      ]
      source_path = tmp_dir + "/test_sliding_group_window_over_time.csv"
      with open(source_path, "w") as fd:
          for ele in data:
              fd.write(ele + "\n")
  
      from pyflink.table.window import Slide
  
      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 SMALLINT,
                  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)
              WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      t.window(
          Slide.over(lit(1).hours)
          .every(lit(30).minutes)
          .on(col("rowtime"))
          .alias("w")
      ).group_by(t.a, t.b, col("w")).select(
          t.a, col("w").start, col("w").end, mean_udaf(t.c).alias("b")
      ).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]",
              "+I[1, 2018-03-11T03:00, 2018-03-11T04:00, 2.5]",
              "+I[1, 2018-03-11T03:30, 2018-03-11T04:30, 5.5]",
              "+I[1, 2018-03-11T04:00, 2018-03-11T05:00, 8.0]",
              "+I[2, 2018-03-11T02:30, 2018-03-11T03:30, 1.0]",
              "+I[2, 2018-03-11T03:00, 2018-03-11T04:00, 2.0]",
              "+I[2, 2018-03-11T03:30, 2018-03-11T04:30, 3.0]",
              "+I[3, 2018-03-11T03:00, 2018-03-11T04:00, 2.0]",
              "+I[3, 2018-03-11T02:30, 2018-03-11T03:30, 2.0]",
          ],
      )
      os.remove(source_path)
  
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 2 for calling the StreamPandasUDAFITTests obj's underlying member method code with example usage: obj.test_sliding_group_window_over_time() without return types