[SPARK-43995][SPARK-43996][CONNECT] Add support for UDFRegistration to the Connect Scala Client

connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -417,6 +417,30 @@ class SparkSession private[sql] (
     range(start, end, step, Option(numPartitions))
   }
 
+  /**
+   * A collection of methods for registering user-defined functions (UDF).
+   *
+   * The following example registers a Scala closure as UDF:
+   * {{{
+   *   sparkSession.udf.register("myUDF", (arg1: Int, arg2: String) => arg2 + arg1)
+   * }}}
+   *
+   * The following example registers a UDF in Java:
+   * {{{
+   *   sparkSession.udf().register("myUDF",
+   *       (Integer arg1, String arg2) -> arg2 + arg1,
+   *       DataTypes.StringType);
+   * }}}
+   *
+   * @note
+   *   The user-defined functions must be deterministic. Due to optimization, duplicate
+   *   invocations may be eliminated or the function may even be invoked more times than it is
+   *   present in the query.
+   *
+   * @since 3.5.0
+   */
+  lazy val udf: UDFRegistration = new UDFRegistration(this)
+
   // scalastyle:off
   // Disable style checker so "implicits" object can start with lowercase i
   /**
@@ -525,6 +549,13 @@ class SparkSession private[sql] (
     client.execute(plan).asScala.toSeq
   }
 
+  private[sql] def registerUdf(udf: proto.CommonInlineUserDefinedFunction): Unit = {
+    val command = proto.Command.newBuilder().setRegisterFunction(udf).build()
+    val plan = proto.Plan.newBuilder().setCommand(command).build()
+
+    client.execute(plan)
+  }
+
   @DeveloperApi
   def execute(extension: com.google.protobuf.Any): Unit = {
     val command = proto.Command.newBuilder().setExtension(extension).build()