add size for column of Map

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

@@ -6,13 +6,20 @@ import org.apache.spark.sql.{Column, functions => sparkFunctions}
 import scala.math.Ordering
 
 trait UnaryFunctions {
-  /** Returns length of array or map.
+  /** Returns length of array
     *
     * apache/spark
     */
   def size[T, A, V[_] : CatalystSizableCollection](column: TypedColumn[T, V[A]]): TypedColumn[T, Int] =
     new TypedColumn[T, Int](implicitly[CatalystSizableCollection[V]].sizeOp(column.untyped))
 
+  /** Returns length of Map
+    *
+    * apache/spark
+    */
+  def size[T, A, B](column: TypedColumn[T, Map[A, B]]): TypedColumn[T, Int] =
+    new TypedColumn[T, Int](sparkFunctions.size(column.untyped))
+
   /** Sorts the input array for the given column in ascending order, according to
     * the natural ordering of the array elements.
     *
@@ -55,6 +62,7 @@ object CatalystSizableCollection {
   implicit def sizableList: CatalystSizableCollection[List] = new CatalystSizableCollection[List] {
     def sizeOp(col: Column): Column = sparkFunctions.size(col)
   }
+
 }
 
 trait CatalystExplodableCollection[V[_]]

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

@@ -42,6 +42,21 @@ class UnaryFunctionsTest extends TypedDatasetSuite {
     check(forAll(prop[X2[Int, Option[Long]]] _))
   }
 
+  test("size on Map") {
+    def prop[A](xs: List[X1[Map[A, A]]])(implicit arb: Arbitrary[Map[A, A]], enc: TypedEncoder[Map[A, A]]): Prop = {
+      val tds = TypedDataset.create(xs)
+
+      val framelessResults = tds.select(size(tds('a))).collect().run().toVector
+      val scalaResults = xs.map(x => x.a.size).toVector
+
+      framelessResults ?= scalaResults
+    }
+
+    check(forAll(prop[Long] _))
+    check(forAll(prop[Int] _))
+    check(forAll(prop[Char] _))
+  }
+
   test("sort in ascending order") {
     def prop[F[X] <: SeqLike[X, F[X]] : CatalystSortableCollection, A: Ordering](xs: List[X1[F[A]]])(implicit enc: TypedEncoder[F[A]]): Prop = {
       val tds = TypedDataset.create(xs)