code: def test_row_time_over_rows_window_aggregate_function(self):
      # create source file path
      import tempfile
      import os
  
      tmp_dir = tempfile.gettempdir()
      data = [
          "1,1,2013-01-01 03:10:00",
          "3,2,2013-01-01 03:10:00",
          "2,1,2013-01-01 03:10:00",
          "1,5,2013-01-01 03:10:00",
          "1,8,2013-01-01 04:20:00",
          "2,3,2013-01-01 03:30:00",
      ]
      source_path = (
          tmp_dir + "/test_over_rows_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,
                  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)
  
      sink_table = generate_random_table_name()
      sink_table_ddl = f"""
          CREATE TABLE {sink_table}(a TINYINT, b FLOAT, c SMALLINT) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
      self.t_env.execute_sql(
          f"""
              insert into {sink_table}
              select a,
               mean_udaf(b)
               over (PARTITION BY a ORDER BY rowtime
               ROWS BETWEEN 1 PRECEDING AND CURRENT ROW),
               max_add_min_udaf(b)
               over (PARTITION BY a ORDER BY rowtime
               ROWS BETWEEN 1 PRECEDING AND CURRENT ROW)
              from {source_table}
          """
      ).wait()
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[1, 1.0, 2]",
              "+I[1, 3.0, 6]",
              "+I[1, 6.5, 13]",
              "+I[2, 1.0, 2]",
              "+I[2, 2.0, 4]",
              "+I[3, 2.0, 4]",
          ],
      )
      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 7 for calling the StreamPandasUDAFITTests obj's underlying member method code with example usage: obj.test_row_time_over_rows_window_aggregate_function() without return types