Better UDF support

README.md

@@ -28,6 +28,7 @@ We have opened a Spark Project Improvement Proposal: [Kotlin support for Apache
     - [Overload Resolution Ambiguity](#overload-resolution-ambiguity)
     - [Tuples](#tuples)
     - [Streaming](#streaming)
+    - [User Defined Functions](#user-defined-functions)
 - [Examples](#examples)
 - [Reporting issues/Support](#reporting-issuessupport)
 - [Code of Conduct](#code-of-conduct)
@@ -275,6 +276,48 @@ withSparkStreaming(batchDuration = Durations.seconds(1), timeout = 10_000) { //
 
 For more information, check the [wiki](https://github.com/JetBrains/kotlin-spark-api/wiki/Streaming).
 
+### User Defined Functions
+
+Spark has a way to call functions from SQL using so-called [UDFs](https://spark.apache.org/docs/latest/sql-ref-functions-udf-scalar.html).
+Using the Scala/Java API from Kotlin is not that obvious, so we decided to add special UDF support for Kotlin.
+This support grew into a typesafe, name-safe, and feature-rich solution for which we will give an example:
+```kotlin
+// example of creation/naming, and registering of a simple UDF
+val plusOne by udf { x: Int -> x + 1 }
+plusOne.register()
+spark.sql("SELECT plusOne(5)").show()
+// +----------+
+// |plusOne(5)|
+// +----------+
+// |         6|
+// +----------+
+
+// directly registering
+udf.register("plusTwo") { x: Double -> x + 2.0 }
+spark.sql("SELECT plusTwo(2.0d)").show()
+// +------------+
+// |plusTwo(2.0)|
+// +------------+
+// |         4.0|
+// +------------+
+
+// dataset select
+val result: Dataset<Int> = myDs.select(
+  plusOne(col(MyType::age))
+)
+```
+
+We support:
+  - a notation close to Spark's
+  - smart naming (with reflection)
+  - creation from function references
+  - typed column operations
+  - UDAF support and functional creation
+  - (Unique!) simple vararg UDF support
+
+For more, check the [extensive examples](examples/src/main/kotlin/org/jetbrains/kotlinx/spark/examples/UDFs.kt).
+Also, check out the [wiki](https://github.com/Kotlin/kotlin-spark-api/wiki/UDF).
+
 ## Examples
 
 For more, check out [examples](examples/src/main/kotlin/org/jetbrains/kotlinx/spark/examples) module.

core/3.2/src/main/scala/org/jetbrains/kotlinx/spark/extensions/VarargUnwrapper.scala

@@ -0,0 +1,93 @@
+package org.jetbrains.kotlinx.spark.extensions
+
+import org.apache.spark.sql.api.java.{UDF1, UDF2}
+
+/**
+ * Allows any simple vararg function reference to be treated as 23 different Scala functions.
+ * Used to make vararg UDFs for `ScalaUDF`.
+ *
+ * @param varargFunc
+ * @param newArray
+ * @tparam T
+ * @tparam Array
+ * @tparam R
+ */
+class VarargUnwrapper[T, Array, R](
+    val varargFunc: UDF1[Array, R],
+    val newArray: UDF2[Integer, UDF1[Integer, T], Array],
+) extends Serializable
+  with Function0[R]
+  with Function1[T, R]
+  with Function2[T, T, R]
+  with Function3[T, T, T, R]
+  with Function4[T, T, T, T, R]
+  with Function5[T, T, T, T, T, R]
+  with Function6[T, T, T, T, T, T, R]
+  with Function7[T, T, T, T, T, T, T, R]
+  with Function8[T, T, T, T, T, T, T, T, R]
+  with Function9[T, T, T, T, T, T, T, T, T, R]
+  with Function10[T, T, T, T, T, T, T, T, T, T, R]
+  with Function11[T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function12[T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function13[T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function14[T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function15[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function16[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function17[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function18[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function19[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function20[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function21[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R]
+  with Function22[T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, T, R] {
+
+  private def vararg(t: T*): R = varargFunc.call(newArray.call(t.size, { t(_) }))
+
+  override def curried: Nothing = throw new UnsupportedOperationException()
+  override def tupled: Nothing = throw new UnsupportedOperationException()
+
+  override def apply(): R = vararg()
+
+  override def apply(v0: T): R = vararg(v0)
+
+  override def apply(v0: T, v1: T): R = vararg(v0, v1)
+
+  override def apply(v0: T, v1: T, v2: T): R = vararg(v0, v1, v2)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T): R = vararg(v0, v1, v2, v3)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T): R = vararg(v0, v1, v2, v3, v4)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T): R = vararg(v0, v1, v2, v3, v4, v5)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T): R = vararg(v0, v1, v2, v3, v4, v5, v6)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T, v17: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T, v17: T, v18: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T, v17: T, v18: T, v19: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18, v19)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T, v17: T, v18: T, v19: T, v20: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18, v19, v20)
+
+  override def apply(v0: T, v1: T, v2: T, v3: T, v4: T, v5: T, v6: T, v7: T, v8: T, v9: T, v10: T, v11: T, v12: T, v13: T, v14: T, v15: T, v16: T, v17: T, v18: T, v19: T, v20: T, v21: T): R = vararg(v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15, v16, v17, v18, v19, v20, v21)
+}

examples/src/main/kotlin/org/jetbrains/kotlinx/spark/examples/Main.kt

@@ -61,7 +61,7 @@ object Main {
         triples
             .leftJoin(
                 right = pairs,
-                col = triples("_1").multiply(2) eq pairs("_1"),
+                col = triples.col("_1").multiply(2) eq pairs.col("_1"),
             )
 //                .also { it.printSchema() }
             .map { (triple, pair) -> Five(triple._1, triple._2, triple._3, pair?._1, pair?._2) }