fix: struct type for nullable field

* fix StructType for nullable list and map field in 3.0

* fix StructType for nullable list and map field in 2.4

* fix test name

Co-authored-by: can wang <can.wang@thoughtworks.com>

Author: nonpool

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

@@ -884,7 +884,7 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
             KComplexTypeWrapper(
                 DataTypes.createArrayType(schema(listParam, types), listParam.isMarkedNullable),
                 klass.java,
-                listParam.isMarkedNullable
+                type.isMarkedNullable
             )
         }
         klass.isSubclassOf(Map::class) -> {
@@ -897,7 +897,7 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
                     true
                 ),
                 klass.java,
-                mapValueParam.isMarkedNullable
+                type.isMarkedNullable
             )
         }
         klass.isData -> {

kotlin-spark-api/2.4/src/test/kotlin/org/jetbrains/kotlinx/spark/api/ApiTest.kt

@@ -542,6 +542,12 @@ class ApiTest : ShouldSpec({
                     .reduceK { a, b -> a + b }
                 expect(result).toBe(3)
             }
+            should("Generate schema correctly with nullalble list and map") {
+                val schema = encoder<NullFieldAbleDataClass>().schema()
+                schema.fields().forEach {
+                    it.nullable() shouldBe true
+                }
+            }
         }
     }
 })
@@ -591,3 +597,9 @@ data class ComplexEnumDataClass(
     val someOtherArray: Array<SomeOtherEnum>,
     val enumMap: Map<SomeEnum, SomeOtherEnum>,
 )
+
+
+data class NullFieldAbleDataClass(
+    val optionList: List<Int>?,
+    val optionMap: Map<String, Int>?
+)

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

@@ -887,7 +887,7 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
             KComplexTypeWrapper(
                 DataTypes.createArrayType(schema(listParam, types), listParam.isMarkedNullable),
                 klass.java,
-                listParam.isMarkedNullable
+                type.isMarkedNullable
             )
         }
         klass.isSubclassOf(Map::class) -> {
@@ -900,7 +900,7 @@ fun schema(type: KType, map: Map<String, KType> = mapOf()): DataType {
                     true
                 ),
                 klass.java,
-                mapValueParam.isMarkedNullable
+                type.isMarkedNullable
             )
         }
         klass.isData -> {

kotlin-spark-api/3.0/src/test/kotlin/org/jetbrains/kotlinx/spark/api/ApiTest.kt

@@ -585,7 +585,12 @@ class ApiTest : ShouldSpec({
                     .reduceK { a, b -> a + b }
                 expect(result).toBe(3)
             }
-
+            should("Generate schema correctly with nullalble list and map") {
+                val schema = encoder<NullFieldAbleDataClass>().schema()
+                schema.fields().forEach {
+                    it.nullable() shouldBe true
+                }
+            }
         }
     }
 })
@@ -615,4 +620,9 @@ data class ComplexEnumDataClass(
     val someEnumArray: Array<SomeEnum>,
     val someOtherArray: Array<SomeOtherEnum>,
     val enumMap: Map<SomeEnum, SomeOtherEnum>,
+)
+
+data class NullFieldAbleDataClass(
+    val optionList: List<Int>?,
+    val optionMap: Map<String, Int>?
 )