adding more working cases, found option cases not to work

Author: Jolanrensen

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

@@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.analysis.GetColumnByOrdinal
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.expressions.{Expression, _}
 import org.apache.spark.sql.catalyst.util.ArrayBasedMapData
-import org.apache.spark.sql.catalyst.{InternalRow, ScalaReflection, WalkedTypePath}
+import org.apache.spark.sql.catalyst.{DefinedByConstructorParams, InternalRow, ScalaReflection, WalkedTypePath}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
 import org.apache.spark.util.Utils
@@ -42,11 +42,12 @@ import java.lang.Exception
  * for classes whose fields are entirely defined by constructor params but should not be
  * case classes.
  */
-trait DefinedByConstructorParams
+//trait DefinedByConstructorParams
 
 /**
  * KotlinReflection is heavily inspired by ScalaReflection and even extends it just to add several methods
  */
+//noinspection RedundantBlock
 object KotlinReflection extends KotlinReflection {
     /**
      * Returns the Spark SQL DataType for a given java class.  Where this is not an exact mapping
@@ -916,9 +917,18 @@ object KotlinReflection extends KotlinReflection {
             }
             //</editor-fold>
 
-            case _ if predefinedDt.isDefined => {
+            // Kotlin specific cases
+            case t if predefinedDt.isDefined => {
+
+//                if (seenTypeSet.contains(t)) {
+//                    throw new UnsupportedOperationException(
+//                        s"cannot have circular references in class, but got the circular reference of class $t"
+//                    )
+//                }
+
                 predefinedDt.get match {
 
+                    // Kotlin data class
                     case dataType: KDataTypeWrapper => {
                         val cls = dataType.cls
                         val properties = getJavaBeanReadableProperties(cls)

core/3.2/src/main/scala/org/jetbrains/kotlinx/spark/extensions/DemoCaseClass.scala

@@ -0,0 +1,3 @@
+package org.jetbrains.kotlinx.spark.extensions
+
+case class DemoCaseClass[T](a: Int, b: T)

kotlin-spark-api/3.2/src/test/kotlin/org/jetbrains/kotlinx/spark/api/EncodingTest.kt

@@ -27,6 +27,7 @@ import org.apache.spark.sql.Dataset
 import org.apache.spark.sql.types.Decimal
 import org.apache.spark.unsafe.types.CalendarInterval
 import org.jetbrains.kotlinx.spark.api.tuples.*
+import org.jetbrains.kotlinx.spark.extensions.DemoCaseClass
 import scala.*
 import java.math.BigDecimal
 import java.sql.Date
@@ -177,13 +178,59 @@ class EncodingTest : ShouldSpec({
     context("schema") {
         withSpark(props = mapOf("spark.sql.codegen.comments" to true)) {
 
-            should("handle Scala case class datasets") {
+            should("handle Scala Case class datasets") {
+                val caseClasses = listOf(
+                    DemoCaseClass(1, "1"),
+                    DemoCaseClass(2, "2"),
+                    DemoCaseClass(3, "3"),
+                )
+                val dataset = caseClasses.toDS()
+                dataset.show()
+                dataset.collectAsList() shouldBe caseClasses
+            }
+
+            should("handle Scala Case class with data class datasets") {
+                val caseClasses = listOf(
+                    DemoCaseClass(1, "1" to 1L),
+                    DemoCaseClass(2, "2" to 2L),
+                    DemoCaseClass(3, "3" to 3L),
+                )
+                val dataset = caseClasses.toDS()
+                dataset.show()
+                dataset.collectAsList() shouldBe caseClasses
+            }
+
+            should("handle data class with Scala Case class datasets") {
+                val caseClasses = listOf(
+                    1 to DemoCaseClass(1, "1"),
+                    2 to DemoCaseClass(2, "2"),
+                    3 to DemoCaseClass(3, "3"),
+                )
+                val dataset = caseClasses.toDS()
+                dataset.show()
+                dataset.collectAsList() shouldBe caseClasses
+            }
+
+            should("handle data class with Scala Case class & deeper datasets") {
+                val caseClasses = listOf(
+                    1 to DemoCaseClass(1, "1" to DemoCaseClass(1, 1.0)),
+                    2 to DemoCaseClass(2, "2" to DemoCaseClass(2, 2.0)),
+                    3 to DemoCaseClass(3, "3" to DemoCaseClass(3, 3.0)),
+                )
+                val dataset = caseClasses.toDS()
+                dataset.show()
+                dataset.collectAsList() shouldBe caseClasses
+            }
+
+
+            should("handle Scala Option datasets") {
                 val caseClasses = listOf(Some(1), Some(2), Some(3))
                 val dataset = caseClasses.toDS()
+                dataset.show()
                 dataset.collectAsList() shouldBe caseClasses
             }
 
-            should("handle Scala case class case class datasets") {
+            should("handle Scala Option Option datasets") {
                 val caseClasses = listOf(
                     Some(Some(1)),
                     Some(Some(2)),
@@ -193,7 +240,7 @@ class EncodingTest : ShouldSpec({
                 dataset.collectAsList() shouldBe caseClasses
             }
 
-            should("handle data class Scala case class datasets") {
+            should("handle data class Scala Option datasets") {
                 val caseClasses = listOf(
                     Some(1) to Some(2),
                     Some(3) to Some(4),
@@ -203,7 +250,7 @@ class EncodingTest : ShouldSpec({
                 dataset.collectAsList() shouldBe caseClasses
             }
 
-            should("handle Scala case class data class datasets") {
+            should("handle Scala Option data class datasets") {
                 val caseClasses = listOf(
                     Some(1 to 2),
                     Some(3 to 4),