fix: 🐛 Fixes incorrect order of fields in data class

We were reading class properties, not constructor params. That lead to
the fact, that sometimes we were reading properties in incorrect
(alphabetical) order.

✅ Closes: #38

Author: asm0dey

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

@@ -674,23 +674,22 @@ object KotlinReflection extends KotlinReflection {
           case dataType: KDataTypeWrapper =>
             val cls = dataType.cls
             val properties = getJavaBeanReadableProperties(cls)
-            val fields = properties.map { prop =>
-
-              val maybeField = dataType.dt.fields.map(_.asInstanceOf[KStructField]).find(it => it.getterName == prop.getReadMethod.getName)
-              if (maybeField.isEmpty)
-                throw new IllegalArgumentException(s"Field ${prop.getName} is not found among available fields, which are: ${dataType.dt.fields.map(_.name).mkString(", ")}")
-              val fieldName = maybeField.get.name
-              val propClass = maybeField.map(it => it.dataType.asInstanceOf[DataTypeWithClass].cls).get
-              val propDt = maybeField.map(it => it.dataType.asInstanceOf[DataTypeWithClass]).get
-
+            val structFields = dataType.dt.fields.map(_.asInstanceOf[KStructField])
+            val fields = structFields.map { structField =>
+              val maybeProp = properties.find(it => it.getReadMethod.getName == structField.getterName)
+              if (maybeProp.isEmpty) throw new IllegalArgumentException(s"Field ${structField.name} is not found among available props, which are: ${properties.map(_.getName).mkString(", ")}")
+              val fieldName = structField.name
+              val propClass = structField.dataType.asInstanceOf[DataTypeWithClass].cls
+              val propDt = structField.dataType.asInstanceOf[DataTypeWithClass]
               val fieldValue = Invoke(
                 inputObject,
-                prop.getReadMethod.getName,
+                maybeProp.get.getReadMethod.getName,
                 inferExternalType(propClass),
                 returnNullable = propDt.nullable
               )
               val newPath = walkedTypePath.recordField(propClass.getName, fieldName)
               (fieldName, serializerFor(fieldValue, getType(propClass), newPath, seenTypeSet, if (propDt.isInstanceOf[ComplexWrapper]) Some(propDt) else None))
+
             }
             createSerializerForObject(inputObject, fields)
 

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

@@ -33,12 +33,15 @@ import java.beans.PropertyDescriptor
 import java.math.BigDecimal
 import java.sql.Date
 import java.sql.Timestamp
+import java.time.Instant
+import java.time.LocalDate
 import java.util.concurrent.ConcurrentHashMap
 import kotlin.reflect.KClass
 import kotlin.reflect.KType
 import kotlin.reflect.full.declaredMemberProperties
 import kotlin.reflect.full.findAnnotation
 import kotlin.reflect.full.isSubclassOf
+import kotlin.reflect.full.primaryConstructor
 import kotlin.reflect.typeOf
 
 @JvmField
@@ -287,20 +290,22 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
                     mapValueParam.isMarkedNullable
             )
         }
-        else -> {
+        klass.isData -> {
             val structType = StructType(
                     klass
-                            .declaredMemberProperties
+                            .primaryConstructor!!
+                            .parameters
                             .filter { it.findAnnotation<Transient>() == null }
                             .map {
-                                val projectedType = types[it.returnType.toString()] ?: it.returnType
-                                val propertyDescriptor = PropertyDescriptor(it.name, klass.java, "is" + it.name.capitalize(), null)
+                                val projectedType = types[it.type.toString()] ?: it.type
+                                val propertyDescriptor = PropertyDescriptor(it.name, klass.java, "is" + it.name?.capitalize(), null)
                                 KStructField(propertyDescriptor.readMethod.name, StructField(it.name, schema(projectedType, types), projectedType.isMarkedNullable, Metadata.empty()))
                             }
                             .toTypedArray()
             )
             KDataTypeWrapper(structType, klass.java, true)
         }
+        else -> throw IllegalArgumentException("$type is unsupported")
     }
 }