code: def test_execute_sql(self):
      t_env = self.t_env
      table_result = t_env.execute_sql(
          "create table tbl"
          "("
          "   a bigint,"
          "   b int,"
          "   c varchar"
          ") with ("
          "  'connector' = 'COLLECTION',"
          "   'is-bounded' = 'false'"
          ")"
      )
      self.assertIsNone(table_result.get_job_client())
      self.assert_equals(
          table_result.get_table_schema().get_field_names(), ["result"]
      )
      self.assertEqual(
          table_result.get_result_kind(), ResultKind.SUCCESS
      )
      table_result.print()
  
      table_result = t_env.execute_sql(
          "alter table tbl set ('k1' = 'a', 'k2' = 'b')"
      )
      self.assertIsNone(table_result.get_job_client())
      self.assert_equals(
          table_result.get_table_schema().get_field_names(), ["result"]
      )
      self.assertEqual(
          table_result.get_result_kind(), ResultKind.SUCCESS
      )
      table_result.print()
  
      sink_table_ddl = """
          CREATE TABLE sinks(k1 BIGINT, k2 INT, c STRING) WITH ('connector'='test-sink')
          """
      t_env.execute_sql(sink_table_ddl)
      table_result = t_env.execute_sql(
          "insert into sinks select * from tbl"
      )
  
      job_execution_result = (
          table_result.get_job_client()
          .get_job_execution_result()
          .result()
      )
      self.assertIsNotNone(job_execution_result.get_job_id())
      self.assert_equals(
          table_result.get_table_schema().get_field_names(),
          ["default_catalog.default_database.sinks"],
      )
      self.assertEqual(
          table_result.get_result_kind(),
          ResultKind.SUCCESS_WITH_CONTENT,
      )
      table_result.print()
  
      table_result = t_env.execute_sql("drop table tbl")
      self.assertIsNone(table_result.get_job_client())
      self.assert_equals(
          table_result.get_table_schema().get_field_names(), ["result"]
      )
      self.assertEqual(
          table_result.get_result_kind(), ResultKind.SUCCESS
      )
      table_result.print()
  
file_path: flink/flink-python/pyflink/table/tests/test_sql.py
imports: import glob
  import os
  import subprocess
  from pyflink.find_flink_home import _find_flink_source_root
  from pyflink.java_gateway import get_gateway
  from pyflink.table import ResultKind, ExplainDetail
  from pyflink.table import expressions
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkStreamTableTestCase, PyFlinkTestCase
label_desc: Use image node_id 3 for calling the StreamSqlTests obj's underlying member method code with example usage: obj.test_execute_sql() without return types