From 322bd202efeb268d7b2ab4ad81ddf293f0984455 Mon Sep 17 00:00:00 2001 From: vicennial Date: Wed, 12 Jul 2023 18:29:49 +0200 Subject: [PATCH] add registration api --- .../apache/spark/sql/UDFRegistration.scala | 652 ++++++++++++++++ .../org/apache/spark/sql/functions.scala | 711 +++++++++++++++++- .../spark/sql/application/ReplE2ESuite.scala | 10 + 3 files changed, 1371 insertions(+), 2 deletions(-) diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/UDFRegistration.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/UDFRegistration.scala index ea68ba042f39b..950407ad0d492 100644 --- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/UDFRegistration.scala +++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/UDFRegistration.scala @@ -17,6 +17,8 @@ package org.apache.spark.sql +import scala.reflect.runtime.universe.TypeTag + import org.apache.spark.internal.Logging import org.apache.spark.sql.expressions.{ScalarUserDefinedFunction, UserDefinedFunction} @@ -66,4 +68,654 @@ class UDFRegistration(session: SparkSession) extends Logging { s"${other.getClass.getSimpleName} is currently unsupported.") } } + + // scalastyle:off line.size.limit + + /* register 0-22 were generated by this script: + (0 to 22).foreach { x => + val params = (1 to x).map(num => s"A$num").mkString(", ") + val typeTags = (1 to x).map(i => s"A$i: TypeTag").foldLeft("RT: TypeTag")(_ + ", " + _) + println(s""" + |/** + | * Registers a deterministic Scala closure of $x arguments as user-defined function (UDF). + | * @tparam RT return type of UDF. + | * @since 3.5.0 + | */ + |def register[$typeTags](name: String, func: ($params) => RT): UserDefinedFunction = { + | register(name, functions.udf(func)) + |}""".stripMargin) + } + */ + + /** + * Registers a deterministic Scala closure of 0 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag](name: String, func: () => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 1 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag, A1: TypeTag](name: String, func: (A1) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 2 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag, A1: TypeTag, A2: TypeTag]( + name: String, + func: (A1, A2) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 3 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag, A1: TypeTag, A2: TypeTag, A3: TypeTag]( + name: String, + func: (A1, A2, A3) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 4 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag, A1: TypeTag, A2: TypeTag, A3: TypeTag, A4: TypeTag]( + name: String, + func: (A1, A2, A3, A4) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 5 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[RT: TypeTag, A1: TypeTag, A2: TypeTag, A3: TypeTag, A4: TypeTag, A5: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 6 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag](name: String, func: (A1, A2, A3, A4, A5, A6) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 7 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 8 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 9 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 10 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 11 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 12 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 13 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13) => RT) + : UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 14 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14) => RT) + : UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 15 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15) => RT) + : UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 16 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16) => RT) + : UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 17 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag]( + name: String, + func: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16, A17) => RT) + : UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 18 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag]( + name: String, + func: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 19 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag]( + name: String, + func: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 20 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag]( + name: String, + func: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 21 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag, + A21: TypeTag]( + name: String, + func: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20, + A21) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + /** + * Registers a deterministic Scala closure of 22 arguments as user-defined function (UDF). + * @tparam RT + * return type of UDF. + * @since 3.5.0 + */ + def register[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag, + A21: TypeTag, + A22: TypeTag]( + name: String, + func: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20, + A21, + A22) => RT): UserDefinedFunction = { + register(name, functions.udf(func)) + } + + // scalastyle:on line.size.limit } diff --git a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala index b0ae4c9752afc..14d3ce8a1e006 100644 --- a/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala +++ b/connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/functions.scala @@ -7656,6 +7656,27 @@ object functions { // scalastyle:off line.size.limit + /* udf 0-22 were generated by this script: + (0 to 22).foreach { x => + val params = (1 to x).map(num => s"A$num").mkString(", ") + val typeTags = (1 to x).map(i => s"A$i: TypeTag").foldLeft("RT: TypeTag")(_ + ", " + _) + val typeTagsInternal = (1 to x).map(i => s"typeTag[A$i]").foldLeft("typeTag[RT]")(_ + ", " + _) + println(s""" + | /** + | * Defines a Scala closure of $x arguments as user-defined function (UDF). The data types are + | * automatically inferred based on the Scala closure's signature. By default the returned UDF is + | * deterministic. To change it to nondeterministic, call the API + | * `UserDefinedFunction.asNondeterministic()`. + | * + | * @group udf_funcs + | * @since 3.4.0 + | */ + | def udf[$typeTags](f: ($params) => RT): UserDefinedFunction = { + | ScalarUserDefinedFunction(f, $typeTagsInternal) + | }""".stripMargin) + } + */ + /** * Defines a Scala closure of 0 arguments as user-defined function (UDF). The data types are * automatically inferred based on the Scala closure's signature. By default the returned UDF is @@ -7678,7 +7699,7 @@ object functions { * @group udf_funcs * @since 3.4.0 */ - def udf[RT: TypeTag, A1: TypeTag](f: A1 => RT): UserDefinedFunction = { + def udf[RT: TypeTag, A1: TypeTag](f: (A1) => RT): UserDefinedFunction = { ScalarUserDefinedFunction(f, typeTag[RT], typeTag[A1]) } @@ -7903,7 +7924,693 @@ object functions { typeTag[A9], typeTag[A10]) } - // scalastyle:off line.size.limit + + /** + * Defines a Scala closure of 11 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11]) + } + + /** + * Defines a Scala closure of 12 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12]) + } + + /** + * Defines a Scala closure of 13 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13]) + } + + /** + * Defines a Scala closure of 14 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag](f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14) => RT) + : UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14]) + } + + /** + * Defines a Scala closure of 15 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag](f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15) => RT) + : UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15]) + } + + /** + * Defines a Scala closure of 16 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16) => RT) + : UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16]) + } + + /** + * Defines a Scala closure of 17 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16, A17) => RT) + : UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17]) + } + + /** + * Defines a Scala closure of 18 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag]( + f: (A1, A2, A3, A4, A5, A6, A7, A8, A9, A10, A11, A12, A13, A14, A15, A16, A17, A18) => RT) + : UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17], + typeTag[A18]) + } + + /** + * Defines a Scala closure of 19 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag]( + f: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17], + typeTag[A18], + typeTag[A19]) + } + + /** + * Defines a Scala closure of 20 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag]( + f: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17], + typeTag[A18], + typeTag[A19], + typeTag[A20]) + } + + /** + * Defines a Scala closure of 21 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag, + A21: TypeTag]( + f: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20, + A21) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17], + typeTag[A18], + typeTag[A19], + typeTag[A20], + typeTag[A21]) + } + + /** + * Defines a Scala closure of 22 arguments as user-defined function (UDF). The data types are + * automatically inferred based on the Scala closure's signature. By default the returned UDF is + * deterministic. To change it to nondeterministic, call the API + * `UserDefinedFunction.asNondeterministic()`. + * + * @group udf_funcs + * @since 3.4.0 + */ + def udf[ + RT: TypeTag, + A1: TypeTag, + A2: TypeTag, + A3: TypeTag, + A4: TypeTag, + A5: TypeTag, + A6: TypeTag, + A7: TypeTag, + A8: TypeTag, + A9: TypeTag, + A10: TypeTag, + A11: TypeTag, + A12: TypeTag, + A13: TypeTag, + A14: TypeTag, + A15: TypeTag, + A16: TypeTag, + A17: TypeTag, + A18: TypeTag, + A19: TypeTag, + A20: TypeTag, + A21: TypeTag, + A22: TypeTag]( + f: ( + A1, + A2, + A3, + A4, + A5, + A6, + A7, + A8, + A9, + A10, + A11, + A12, + A13, + A14, + A15, + A16, + A17, + A18, + A19, + A20, + A21, + A22) => RT): UserDefinedFunction = { + ScalarUserDefinedFunction( + f, + typeTag[RT], + typeTag[A1], + typeTag[A2], + typeTag[A3], + typeTag[A4], + typeTag[A5], + typeTag[A6], + typeTag[A7], + typeTag[A8], + typeTag[A9], + typeTag[A10], + typeTag[A11], + typeTag[A12], + typeTag[A13], + typeTag[A14], + typeTag[A15], + typeTag[A16], + typeTag[A17], + typeTag[A18], + typeTag[A19], + typeTag[A20], + typeTag[A21], + typeTag[A22]) + } + + // scalastyle:on line.size.limit /** * Call a builtin or temp function. diff --git a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala index d6404067c66b1..c0f09ad44c1d3 100644 --- a/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala +++ b/connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/application/ReplE2ESuite.scala @@ -217,4 +217,14 @@ class ReplE2ESuite extends RemoteSparkSession with BeforeAndAfterEach { val output = runCommandsInShell(input) assertContains("Array[Long] = Array(0L, 100L, 200L, 300L, 400L)", output) } + + test("UDF closure registration") { + val input = """ + |class A(x: Int) { def get = x * 15 } + |spark.udf.register("directUdf", (x: Int) => new A(x).get) + |spark.sql("select directUdf(id) from range(5)").as[Long].collect() + """.stripMargin + val output = runCommandsInShell(input) + assertContains("Array[Long] = Array(0L, 15L, 30L, 45L, 60L)", output) + } }