code: def test_flat_map(self):
      t = self.t_env.from_elements(
          [(1, "2,3"), (2, "1"), (1, "5,6,7")],
          DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.TINYINT()),
                  DataTypes.FIELD("b", DataTypes.STRING()),
              ]
          ),
      )
  
      sink_table_ddl = """
              CREATE TABLE Results_test_flat_map(
                  a BIGINT, b STRING, c BIGINT, d STRING, e BIGINT, f STRING
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
  
      @udtf(result_types=[DataTypes.INT(), DataTypes.STRING()])
      def split(x):
          for s in x.b.split(","):
              yield x.a, s
  
      t.flat_map(split).alias("a", "b").flat_map(split).alias(
          "a", "b"
      ).join_lateral(split.alias("c", "d")).left_outer_join_lateral(
          split.alias("e", "f")
      ).execute_insert(
          "Results_test_flat_map"
      ).wait()
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[1, 2, 1, 2, 1, 2]",
              "+I[1, 3, 1, 3, 1, 3]",
              "+I[2, 1, 2, 1, 2, 1]",
              "+I[1, 5, 1, 5, 1, 5]",
              "+I[1, 6, 1, 6, 1, 6]",
              "+I[1, 7, 1, 7, 1, 7]",
          ],
      )
  
imports: import unittest
  from pandas.testing import assert_frame_equal
  from pyflink.common import Row
  from pyflink.table import expressions, ListView
  from pyflink.table.types import DataTypes
  from pyflink.table.udf import udf, udtf, udaf, AggregateFunction, TableAggregateFunction, udtaf
  from pyflink.testing import source_sink_utils
  from pyflink.testing.test_case_utils import PyFlinkBatchTableTestCase, PyFlinkStreamTableTestCase
label_desc: Use image node_id 3 for calling the RowBasedOperationTests obj's underlying member method code with example usage: obj.test_flat_map() without return types