code: def test_map_with_pandas_udf(self):
      t = self.t_env.from_elements(
          [
              (1, Row(2, 3)),
              (2, Row(1, 3)),
              (1, Row(5, 4)),
              (1, Row(8, 6)),
              (2, Row(3, 4)),
          ],
          DataTypes.ROW(
              [
                  DataTypes.FIELD("a", DataTypes.TINYINT()),
                  DataTypes.FIELD(
                      "b",
                      DataTypes.ROW(
                          [
                              DataTypes.FIELD("c", DataTypes.INT()),
                              DataTypes.FIELD("d", DataTypes.INT()),
                          ]
                      ),
                  ),
              ]
          ),
      )
  
      sink_table_ddl = """
              CREATE TABLE Results_test_map_with_pandas_udf(
                  a BIGINT,
                  b BIGINT
              ) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
  
      def func(x):
          import pandas as pd
  
          res = pd.concat([x.a, x.c + x.d], axis=1)
          return res
  
      def func2(x):
          return x * 2
  
      def func3(x):
          assert isinstance(x, Row)
          return x
  
      pandas_udf = udf(
          func,
          result_type=DataTypes.ROW(
              [
                  DataTypes.FIELD("c", DataTypes.BIGINT()),
                  DataTypes.FIELD("d", DataTypes.BIGINT()),
              ]
          ),
          func_type="pandas",
      )
  
      pandas_udf_2 = udf(
          func2,
          result_type=DataTypes.ROW(
              [
                  DataTypes.FIELD("c", DataTypes.BIGINT()),
                  DataTypes.FIELD("d", DataTypes.BIGINT()),
              ]
          ),
          func_type="pandas",
      )
  
      general_udf = udf(
          func3,
          result_type=DataTypes.ROW(
              [
                  DataTypes.FIELD("c", DataTypes.BIGINT()),
                  DataTypes.FIELD("d", DataTypes.BIGINT()),
              ]
          ),
      )
  
      t.map(pandas_udf).map(pandas_udf_2).map(
          general_udf
      ).execute_insert("Results_test_map_with_pandas_udf").wait()
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[4, 8]",
              "+I[2, 10]",
              "+I[2, 28]",
              "+I[2, 18]",
              "+I[4, 14]",
          ],
      )
  
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 2 for calling the RowBasedOperationTests obj's underlying member method code with example usage: obj.test_map_with_pandas_udf() and returns: res, unknown, x