code: def test_execute_over_aggregate_from_json_plan(self):
      # create source file path
      tmp_dir = self.tempdir
      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_execute_over_aggregate_from_json_plan.csv"
      )
      sink_path = (
          tmp_dir + "/test_execute_over_aggregate_from_json_plan"
      )
      with open(source_path, "w") as fd:
          for ele in data:
              fd.write(ele + "\n")
  
      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' = 'filesystem',
                  'path' = '{source_path}',
                  'format' = 'csv'
              )
          """
      self.t_env.execute_sql(source_table_ddl)
  
      sink_table = generate_random_table_name()
      self.t_env.execute_sql(
          f"""
              CREATE TABLE {sink_table} (
                  a TINYINT,
                  b FLOAT,
                  c SMALLINT
              ) WITH (
                  'connector' = 'filesystem',
                  'path' = '{sink_path}',
                  'format' = 'csv'
              )
          """
      )
  
      self.t_env.get_config().set(
          "pipeline.time-characteristic", "EventTime"
      )
  
      json_plan = self.t_env._j_tenv.compilePlanSql(
          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}
          """
      )
      from py4j.java_gateway import get_method
  
      get_method(json_plan.execute(), "await")()
  
      import glob
  
      lines = [
          line.strip()
          for file in glob.glob(sink_path + "/*")
          for line in open(file, "r")
      ]
      lines.sort()
      self.assertEqual(
          lines,
          [
              "1,1.0,2",
              "1,3.0,6",
              "1,6.5,13",
              "2,1.0,2",
              "2,2.0,4",
              "3,2.0,4",
          ],
      )
  
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 9 for calling the StreamPandasUDAFITTests obj's underlying member method code with example usage: obj.test_execute_over_aggregate_from_json_plan() without return types