Add Column unbase64

Author: Avasil

dataset/src/main/scala/frameless/functions/NonAggregateFunctions.scala

@@ -233,6 +233,14 @@ trait NonAggregateFunctions {
   def base64[T](column: AbstractTypedColumn[T, Array[Byte]]): column.ThisType[T, String] =
     column.typed(sparkFunctions.base64(column.untyped))
 
+  /** Non-Aggregate function: Decodes a BASE64 encoded string column and returns it as a binary column.
+    * This is the reverse of base64.
+    *
+    * apache/spark
+    */
+  def unbase64[T](column: AbstractTypedColumn[T, String]): column.ThisType[T, Array[Byte]] =
+    column.typed(sparkFunctions.unbase64(column.untyped))
+
   /** Non-Aggregate function: Concatenates multiple input string columns together into a single string column.
     * @note varargs make it harder to generalize so we overload the method for [[TypedColumn]] and [[TypedAggregate]]
     *

dataset/src/test/scala/frameless/functions/NonAggregateFunctionsTests.scala

@@ -536,7 +536,37 @@ class NonAggregateFunctionsTests extends TypedDatasetSuite {
         .run()
         .toList
 
-      res ?= resCompare
+      val backAndForth = typedDS
+        .select(base64(unbase64(base64(typedDS('a)))))
+        .collect()
+        .run()
+        .toList
+
+      (res ?= resCompare) && (res ?= backAndForth)
+    }
+
+    check(forAll(prop _))
+  }
+
+  test("unbase64") {
+    val spark = session
+    import spark.implicits._
+
+    def prop(values: List[X1[String]])(implicit encX1: Encoder[X1[String]]) = {
+      val cDS = session.createDataset(values)
+      val resCompare = cDS
+        .select(sparkFunctions.unbase64(cDS("a")))
+        .map(_.getAs[Array[Byte]](0))
+        .collect().toList
+
+      val typedDS = TypedDataset.create(values)
+      val res = typedDS
+        .select(unbase64(typedDS('a)))
+        .collect()
+        .run()
+        .toList
+
+      res.map(_.toList) ?= resCompare.map(_.toList)
     }
 
     check(forAll(prop _))