code: def test_execute_from_json_plan(self):
      # create source file path
      tmp_dir = self.tempdir
      data = ["1,1", "3,2", "2,1"]
      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 (
                  a BIGINT,
                  b BIGINT,
                  c BIGINT
              ) WITH (
                  'connector' = 'filesystem',
                  'path' = '%s',
                  'format' = 'csv'
              )
          """
          % sink_path
      )
  
      self.t_env.create_temporary_system_function(
          "multi_emit2",
          udtf(
              MultiEmit(),
              result_types=[DataTypes.BIGINT(), DataTypes.BIGINT()],
          ),
      )
  
      json_plan = self.t_env._j_tenv.compilePlanSql(
          "INSERT INTO sink_table "
          "SELECT a, x, y FROM source_table "
          "LEFT JOIN LATERAL TABLE(multi_emit2(a, b))"
          " as T(x, y)"
          " ON TRUE"
      )
      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,2,0", "3,3,0", "3,3,1"])
  
imports: import unittest
  from pyflink.table import DataTypes
  from pyflink.table.udf import TableFunction, udtf, ScalarFunction, udf
  from pyflink.table.expressions import col
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, PyFlinkBatchTableTestCase
label_desc: Use image node_id 1 for calling the PyFlinkStreamUserDefinedFunctionTests obj's underlying member method code with example usage: obj.test_execute_from_json_plan() without return types