feat: adds support for date encoding

Author: Jolanrensen

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

@@ -364,6 +364,8 @@ private val knownDataTypes = mapOf(
         Double::class to DataTypes.DoubleType,
         String::class to DataTypes.StringType,
         LocalDate::class to `DateType$`.`MODULE$`,
+        Date::class to `DateType$`.`MODULE$`,
+        Timestamp::class to `TimestampType$`.`MODULE$`,
         Instant::class to `TimestampType$`.`MODULE$`
 )
 

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

@@ -21,7 +21,10 @@ import ch.tutteli.atrium.api.fluent.en_GB.*
 import ch.tutteli.atrium.domain.builders.migration.asExpect
 import ch.tutteli.atrium.verbs.expect
 import io.kotest.core.spec.style.ShouldSpec
+import org.apache.spark.sql.Dataset
 import java.io.Serializable
+import java.sql.Date
+import java.sql.Timestamp
 import java.time.LocalDate
 
 class ApiTest : ShouldSpec({
@@ -156,6 +159,18 @@ class ApiTest : ShouldSpec({
 
                 expect(result).asExpect().contains.inOrder.only.values(3, 5, 7, 9, 11)
             }
+            should("be able to serialize Date 2.4") { // uses knownDataTypes
+                val dataset: Dataset<Pair<Date, Int>> = dsOf(Date.valueOf("2020-02-10") to 5)
+                dataset.show()
+            }
+            should("handle Timestamp Datasets 2.4") { // uses encoder
+                val dataset = dsOf(Timestamp(0L))
+                dataset.show()
+            }
+            should("be able to serialize Timestamp 2.4") { // uses knownDataTypes
+                val dataset = dsOf(Timestamp(0L) to 2)
+                dataset.show()
+            }
         }
     }
 })

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

@@ -56,7 +56,9 @@ val ENCODERS = mapOf<KClass<*>, Encoder<*>>(
         String::class to STRING(),
         BigDecimal::class to DECIMAL(),
         Date::class to DATE(),
+        LocalDate::class to LOCALDATE(), // 3.0 only
         Timestamp::class to TIMESTAMP(),
+        Instant::class to INSTANT(), // 3.0 only
         ByteArray::class to BINARY()
 )
 
@@ -351,6 +353,8 @@ private val knownDataTypes = mapOf(
         Double::class to DataTypes.DoubleType,
         String::class to DataTypes.StringType,
         LocalDate::class to `DateType$`.`MODULE$`,
+        Date::class to `DateType$`.`MODULE$`,
+        Timestamp::class to `TimestampType$`.`MODULE$`,
         Instant::class to `TimestampType$`.`MODULE$`
 )
 

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

@@ -21,7 +21,11 @@ import ch.tutteli.atrium.api.fluent.en_GB.*
 import ch.tutteli.atrium.domain.builders.migration.asExpect
 import ch.tutteli.atrium.verbs.expect
 import io.kotest.core.spec.style.ShouldSpec
+import org.apache.spark.sql.Dataset
 import java.io.Serializable
+import java.sql.Date
+import java.sql.Timestamp
+import java.time.Instant
 import java.time.LocalDate
 
 class ApiTest : ShouldSpec({
@@ -169,6 +173,26 @@ class ApiTest : ShouldSpec({
 
                 expect(result).asExpect().contains.inOrder.only.values(3, 5, 7, 9, 11)
             }
+            should("handle LocalDate Datasets") { // uses encoder
+                val dataset: Dataset<LocalDate> = dsOf(LocalDate.now(), LocalDate.now())
+                dataset.show()
+            }
+            should("handle Instant Datasets") { // uses encoder
+                val dataset: Dataset<Instant> = dsOf(Instant.now(), Instant.now())
+                dataset.show()
+            }
+            should("be able to serialize Date") { // uses knownDataTypes
+                val dataset: Dataset<Pair<Date, Int>> = dsOf(Date.valueOf("2020-02-10") to 5)
+                dataset.show()
+            }
+            should("handle Timestamp Datasets") { // uses encoder
+                val dataset = dsOf(Timestamp(0L))
+                dataset.show()
+            }
+            should("be able to serialize Timestamp") { // uses knownDataTypes
+                val dataset = dsOf(Timestamp(0L) to 2)
+                dataset.show()
+            }
         }
     }
 })