code: def test_scalar_function(self):
      # test metric disabled.
      self.t_env.get_config().set("python.metric.enabled", "false")
      self.t_env.get_config().set(
          "pipeline.global-job-parameters", "subtract_value:2"
      )
      # test lambda function
      add_one = udf(lambda i: i + 1, result_type=DataTypes.BIGINT())
  
      # test Python ScalarFunction
      subtract_one = udf(SubtractOne(), result_type=DataTypes.BIGINT())
  
      subtract_two = udf(
          SubtractWithParameters(), result_type=DataTypes.BIGINT()
      )
  
      # test callable function
      add_one_callable = udf(
          CallablePlus(), result_type=DataTypes.BIGINT()
      )
  
      def partial_func(col, param):
          return col + param
  
      # test partial function
      import functools
  
      add_one_partial = udf(
          functools.partial(partial_func, param=1),
          result_type=DataTypes.BIGINT(),
      )
  
      # check memory limit is set
      @udf(result_type=DataTypes.BIGINT())
      def check_memory_limit(exec_mode):
          if exec_mode == "process":
              assert (
                  os.environ["_PYTHON_WORKER_MEMORY_LIMIT"] is not None
              )
          return 1
  
      sink_table = generate_random_table_name()
      sink_table_ddl = f"""
              CREATE TABLE {sink_table}(a BIGINT, b BIGINT, c BIGINT, d BIGINT, e BIGINT, f BIGINT,
               g BIGINT, h BIGINT) WITH ('connector'='test-sink')
          """
      self.t_env.execute_sql(sink_table_ddl)
  
      execution_mode = self.t_env.get_config().get(
          "python.execution-mode", "process"
      )
  
      t = self.t_env.from_elements(
          [(1, 2, 3), (2, 5, 6), (3, 1, 9)], ["a", "b", "c"]
      )
      t.where(add_one(t.b) <= 3).select(
          add_one(t.a),
          subtract_one(t.b),
          subtract_two(t.b),
          add(t.a, t.c),
          add_one_callable(t.a),
          add_one_partial(t.a),
          check_memory_limit(execution_mode),
          t.a,
      ).execute_insert(sink_table).wait()
      actual = source_sink_utils.results()
      self.assert_equals(
          actual,
          [
              "+I[2, 1, 0, 4, 2, 2, 1, 1]",
              "+I[4, 0, -1, 12, 4, 4, 1, 3]",
          ],
      )
  
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 1 for calling the UserDefinedFunctionTests obj's underlying member method code with example usage: obj.test_scalar_function() and returns: unknown, int