code: def test_execute_from_json_plan(self):
      # create source file path
      tmp_dir = self.tempdir
      data = ["1,1", "3,3", "2,2"]
      source_path = tmp_dir + "/test_execute_from_json_plan_input.csv"
      sink_path = tmp_dir + "/test_execute_from_json_plan_out"
      with open(source_path, "w") as fd:
          for ele in data:
              fd.write(ele + "\n")
  
      source_table = (
          """
              CREATE TABLE source_table (
                  a BIGINT,
                  b BIGINT
              ) WITH (
                  'connector' = 'filesystem',
                  'path' = '%s',
                  'format' = 'csv'
              )
          """
          % source_path
      )
      self.t_env.execute_sql(source_table)
  
      self.t_env.execute_sql(
          """
              CREATE TABLE sink_table (
                  id BIGINT,
                  data BIGINT
              ) WITH (
                  'connector' = 'filesystem',
                  'path' = '%s',
                  'format' = 'csv'
              )
          """
          % sink_path
      )
  
      add_one = udf(lambda i: i + 1, result_type=DataTypes.BIGINT())
      self.t_env.create_temporary_system_function("add_one", add_one)
  
      json_plan = self.t_env._j_tenv.compilePlanSql(
          "INSERT INTO sink_table SELECT "
          "a, "
          "add_one(b) "
          "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,2", "2,3", "3,4"])
  
imports: import datetime
  import os
  import unittest
  import uuid
  import pytz
  from pyflink.common import Row
  from pyflink.table import DataTypes, expressions
  from pyflink.table.expressions import call
  from pyflink.table.udf import ScalarFunction, udf, FunctionContext
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, PyFlinkBatchTableTestCase
label_desc: Use image node_id 6 for calling the PyFlinkStreamUserDefinedFunctionTests obj's underlying member method code with example usage: obj.test_execute_from_json_plan() without return types