From 56bc82f9ff0de6948386e8c4a330ba715285ad55 Mon Sep 17 00:00:00 2001
From: Pasha Finkelshteyn <asm0dey@jetbrains.com>
Date: Wed, 5 Aug 2020 20:41:48 +0300
Subject: [PATCH] =?UTF-8?q?fix:=20=F0=9F=90=9B=20Fixes=20incorrect=20order?=
 =?UTF-8?q?=20of=20fields=20in=20data=20class?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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
---
 .../apache/spark/sql/KotlinReflection.scala   | 19 +++++++++----------
 .../kotlin/org/jetbrains/spark/api/ApiV1.kt   | 13 +++++++++----
 2 files changed, 18 insertions(+), 14 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/sql/KotlinReflection.scala b/core/src/main/scala/org/apache/spark/sql/KotlinReflection.scala
index 2767fa42..8e2b5e89 100644
--- a/core/src/main/scala/org/apache/spark/sql/KotlinReflection.scala
+++ b/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)
 
diff --git a/kotlin-spark-api/src/main/kotlin/org/jetbrains/spark/api/ApiV1.kt b/kotlin-spark-api/src/main/kotlin/org/jetbrains/spark/api/ApiV1.kt
index aecca01d..d0119b60 100644
--- a/kotlin-spark-api/src/main/kotlin/org/jetbrains/spark/api/ApiV1.kt
+++ b/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")
     }
 }