fix: 🐛 Fixes incorrect handling of primitive types

There was missing primitive types handling in `KotlinReflection`

✅ Closes: #37

Author: asm0dey

core/src/main/scala/org/apache/spark/sql/KotlinReflection.scala

@@ -213,26 +213,44 @@ object KotlinReflection extends KotlinReflection {
       case t if isSubtype(t, localTypeOf[java.lang.Long]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Long])
+      case t if isSubtype(t, localTypeOf[Long]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Long])
 
       case t if isSubtype(t, localTypeOf[java.lang.Double]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Double])
+      case t if isSubtype(t, localTypeOf[Double]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Double])
 
       case t if isSubtype(t, localTypeOf[java.lang.Float]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Float])
+      case t if isSubtype(t, localTypeOf[Float]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Float])
 
       case t if isSubtype(t, localTypeOf[java.lang.Short]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Short])
+      case t if isSubtype(t, localTypeOf[Short]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Short])
 
       case t if isSubtype(t, localTypeOf[java.lang.Byte]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Byte])
+      case t if isSubtype(t, localTypeOf[Byte]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Byte])
 
       case t if isSubtype(t, localTypeOf[java.lang.Boolean]) =>
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Boolean])
+      case t if isSubtype(t, localTypeOf[Boolean]) =>
+        createDeserializerForTypesSupportValueOf(path,
+          classOf[java.lang.Boolean])
 
       case t if isSubtype(t, localTypeOf[java.time.LocalDate]) =>
         createDeserializerForLocalDate(path)
@@ -600,13 +618,17 @@ object KotlinReflection extends KotlinReflection {
       case t if isSubtype(t, localTypeOf[Int]) =>
         createSerializerForInteger(inputObject)
       case t if isSubtype(t, localTypeOf[java.lang.Long]) => createSerializerForLong(inputObject)
-      case t if isSubtype(t, localTypeOf[java.lang.Double]) =>
-        createSerializerForDouble(inputObject)
+      case t if isSubtype(t, localTypeOf[Long]) => createSerializerForLong(inputObject)
+      case t if isSubtype(t, localTypeOf[java.lang.Double]) => createSerializerForDouble(inputObject)
+      case t if isSubtype(t, localTypeOf[Double]) => createSerializerForDouble(inputObject)
       case t if isSubtype(t, localTypeOf[java.lang.Float]) => createSerializerForFloat(inputObject)
+      case t if isSubtype(t, localTypeOf[Float]) => createSerializerForFloat(inputObject)
       case t if isSubtype(t, localTypeOf[java.lang.Short]) => createSerializerForShort(inputObject)
+      case t if isSubtype(t, localTypeOf[Short]) => createSerializerForShort(inputObject)
       case t if isSubtype(t, localTypeOf[java.lang.Byte]) => createSerializerForByte(inputObject)
-      case t if isSubtype(t, localTypeOf[java.lang.Boolean]) =>
-        createSerializerForBoolean(inputObject)
+      case t if isSubtype(t, localTypeOf[Byte]) => createSerializerForByte(inputObject)
+      case t if isSubtype(t, localTypeOf[java.lang.Boolean]) => createSerializerForBoolean(inputObject)
+      case t if isSubtype(t, localTypeOf[Boolean]) => createSerializerForBoolean(inputObject)
 
       case t if t.typeSymbol.annotations.exists(_.tree.tpe =:= typeOf[SQLUserDefinedType]) =>
         val udt = getClassFromType(t)

kotlin-spark-api/src/main/kotlin/org/jetbrains/spark/api/ApiV1.kt

@@ -320,7 +320,9 @@ private val knownDataTypes = mapOf(
         Boolean::class to DataTypes.BooleanType,
         Float::class to DataTypes.FloatType,
         Double::class to DataTypes.DoubleType,
-        String::class to DataTypes.StringType
+        String::class to DataTypes.StringType,
+        LocalDate::class to `DateType$`.`MODULE$`,
+        Instant::class to `TimestampType$`.`MODULE$`
 )
 
 private fun transitiveMerge(a: Map<String, KType>, b: Map<String, KType>): Map<String, KType> {