[SPARK-27008][SQL] Support java.time.LocalDate as an external type of DateType

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala

@@ -21,7 +21,7 @@ import java.lang.{Iterable => JavaIterable}
 import java.math.{BigDecimal => JavaBigDecimal}
 import java.math.{BigInteger => JavaBigInteger}
 import java.sql.{Date, Timestamp}
-import java.time.Instant
+import java.time.{Instant, LocalDate}
 import java.util.{Map => JavaMap}
 import javax.annotation.Nullable
 
@@ -62,6 +62,7 @@ object CatalystTypeConverters {
       case mapType: MapType => MapConverter(mapType.keyType, mapType.valueType)
       case structType: StructType => StructConverter(structType)
       case StringType => StringConverter
+      case DateType if SQLConf.get.dateExternalType == "LocalDate" => LocalDateConverter
       case DateType => DateConverter
       case TimestampType if SQLConf.get.timestampExternalType == "Instant" => InstantConverter
       case TimestampType => TimestampConverter
@@ -308,6 +309,18 @@ object CatalystTypeConverters {
       DateTimeUtils.toJavaDate(row.getInt(column))
   }
 
+  private object LocalDateConverter extends CatalystTypeConverter[LocalDate, LocalDate, Any] {
+    override def toCatalystImpl(scalaValue: LocalDate): Int = {
+      DateTimeUtils.localDateToDays(scalaValue)
+    }
+    override def toScala(catalystValue: Any): LocalDate = {
+      if (catalystValue == null) null
+      else DateTimeUtils.daysToLocalDate(catalystValue.asInstanceOf[Int])
+    }
+    override def toScalaImpl(row: InternalRow, column: Int): LocalDate =
+      DateTimeUtils.daysToLocalDate(row.getInt(column))
+  }
+
   private object TimestampConverter extends CatalystTypeConverter[Timestamp, Timestamp, Any] {
     override def toCatalystImpl(scalaValue: Timestamp): Long =
       DateTimeUtils.fromJavaTimestamp(scalaValue)
@@ -433,6 +446,7 @@ object CatalystTypeConverters {
   def convertToCatalyst(a: Any): Any = a match {
     case s: String => StringConverter.toCatalyst(s)
     case d: Date => DateConverter.toCatalyst(d)
+    case ld: LocalDate => LocalDateConverter.toCatalyst(ld)
     case t: Timestamp => TimestampConverter.toCatalyst(t)
     case i: Instant => InstantConverter.toCatalyst(i)
     case d: BigDecimal => new DecimalConverter(DecimalType(d.precision, d.scale)).toCatalyst(d)

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/DeserializerBuildHelper.scala

@@ -101,6 +101,15 @@ object DeserializerBuildHelper {
       returnNullable = false)
   }
 
+  def createDeserializerForLocalDate(path: Expression): Expression = {
+    StaticInvoke(
+      DateTimeUtils.getClass,
+      ObjectType(classOf[java.time.LocalDate]),
+      "daysToLocalDate",
+      path :: Nil,
+      returnNullable = false)
+  }
+
   def createDeserializerForInstant(path: Expression): Expression = {
     StaticInvoke(
       DateTimeUtils.getClass,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/JavaTypeInference.scala

@@ -31,6 +31,7 @@ import org.apache.spark.sql.catalyst.analysis.GetColumnByOrdinal
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.objects._
 import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericArrayData}
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 
@@ -221,6 +222,9 @@ object JavaTypeInference {
                 c == classOf[java.lang.Boolean] =>
         createDeserializerForTypesSupportValueOf(path, c)
 
+      case c if c == classOf[java.time.LocalDate] =>
+        createDeserializerForLocalDate(path)
+
       case c if c == classOf[java.sql.Date] =>
         createDeserializerForSqlDate(path)
 
@@ -393,6 +397,14 @@ object JavaTypeInference {
             inputObject :: Nil,
             returnNullable = false)
 
+        case c if c == classOf[java.sql.Date] && SQLConf.get.dateExternalType == "LocalDate" =>
+          StaticInvoke(
+            DateTimeUtils.getClass,
+            DateType,
+            "localDateToDays",
+            inputObject :: Nil,
+            returnNullable = false)
+
         case c if c == classOf[java.sql.Date] =>
           StaticInvoke(
             DateTimeUtils.getClass,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -194,6 +194,9 @@ object ScalaReflection extends ScalaReflection {
         createDeserializerForTypesSupportValueOf(path,
           classOf[java.lang.Boolean])
 
+      case t if t <:< localTypeOf[java.time.LocalDate] =>
+        createDeserializerForLocalDate(path)
+
       case t if t <:< localTypeOf[java.sql.Date] =>
         createDeserializerForSqlDate(path)
 
@@ -493,6 +496,14 @@ object ScalaReflection extends ScalaReflection {
           inputObject :: Nil,
           returnNullable = false)
 
+      case t if t <:< localTypeOf[java.time.LocalDate] =>
+        StaticInvoke(
+          DateTimeUtils.getClass,
+          DateType,
+          "localDateToDays",
+          inputObject :: Nil,
+          returnNullable = false)
+
       case t if t <:< localTypeOf[java.sql.Date] =>
         StaticInvoke(
           DateTimeUtils.getClass,
@@ -704,6 +715,7 @@ object ScalaReflection extends ScalaReflection {
       case t if t <:< localTypeOf[String] => Schema(StringType, nullable = true)
       case t if t <:< localTypeOf[java.time.Instant] => Schema(TimestampType, nullable = true)
       case t if t <:< localTypeOf[java.sql.Timestamp] => Schema(TimestampType, nullable = true)
+      case t if t <:< localTypeOf[java.time.LocalDate] => Schema(DateType, nullable = true)
       case t if t <:< localTypeOf[java.sql.Date] => Schema(DateType, nullable = true)
       case t if t <:< localTypeOf[BigDecimal] => Schema(DecimalType.SYSTEM_DEFAULT, nullable = true)
       case t if t <:< localTypeOf[java.math.BigDecimal] =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/RowEncoder.scala

@@ -46,7 +46,9 @@ import org.apache.spark.unsafe.types.UTF8String
  *   StringType -> String
  *   DecimalType -> java.math.BigDecimal or scala.math.BigDecimal or Decimal
  *
- *   DateType -> java.sql.Date
+ *   DateType -> java.sql.Date when spark.sql.catalyst.dateType is set to Date
+ *   DateType -> java.time.LocalDate when spark.sql.catalyst.dateType is set to LocalDate
+ *
  *   TimestampType -> java.sql.Timestamp when spark.sql.catalyst.timestampType is set to Timestamp
  *   TimestampType -> java.time.Instant when spark.sql.catalyst.timestampType is set to Instant
  *
@@ -107,6 +109,14 @@ object RowEncoder {
         inputObject :: Nil,
         returnNullable = false)
 
+    case DateType if SQLConf.get.dateExternalType == "LocalDate" =>
+      StaticInvoke(
+        DateTimeUtils.getClass,
+        DateType,
+        "localDateToDays",
+        inputObject :: Nil,
+        returnNullable = false)
+
     case DateType =>
       StaticInvoke(
         DateTimeUtils.getClass,
@@ -237,6 +247,8 @@ object RowEncoder {
     case TimestampType if SQLConf.get.timestampExternalType == "Instant" =>
       ObjectType(classOf[java.time.Instant])
     case TimestampType => ObjectType(classOf[java.sql.Timestamp])
+    case DateType if SQLConf.get.dateExternalType == "LocalDate" =>
+      ObjectType(classOf[java.time.LocalDate])
     case DateType => ObjectType(classOf[java.sql.Date])
     case _: DecimalType => ObjectType(classOf[java.math.BigDecimal])
     case StringType => ObjectType(classOf[java.lang.String])
@@ -295,6 +307,14 @@ object RowEncoder {
         input :: Nil,
         returnNullable = false)
 
+    case DateType if SQLConf.get.dateExternalType == "LocalDate" =>
+      StaticInvoke(
+        DateTimeUtils.getClass,
+        ObjectType(classOf[java.time.LocalDate]),
+        "daysToLocalDate",
+        input :: Nil,
+        returnNullable = false)
+
     case DateType =>
       StaticInvoke(
         DateTimeUtils.getClass,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala

@@ -367,6 +367,10 @@ object DateTimeUtils {
     days.toInt
   }
 
+  def localDateToDays(localDate: LocalDate): Int = localDate.toEpochDay.toInt
+
+  def daysToLocalDate(days: Int): LocalDate = LocalDate.ofEpochDay(days)
+
   /**
    * Trim and parse a given UTF8 date string to a corresponding [[Int]] value.
    * The return type is [[Option]] in order to distinguish between 0 and null. The following

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1688,6 +1688,12 @@ object SQLConf {
     .stringConf
     .checkValues(Set("Timestamp", "Instant"))
     .createWithDefault("Timestamp")
+
+  val DATE_EXTERNAL_TYPE = buildConf("spark.sql.catalyst.dateType")
+    .doc("Java class to/from which an instance of DateType is converted.")
+    .stringConf
+    .checkValues(Set("Date", "LocalDate"))
+    .createWithDefault("Date")
 }
 
 /**
@@ -1877,6 +1883,8 @@ class SQLConf extends Serializable with Logging {
 
   def timestampExternalType: String = getConf(TIMESTAMP_EXTERNAL_TYPE)
 
+  def dateExternalType: String = getConf(DATE_EXTERNAL_TYPE)
+
   /**
    * Returns the [[Resolver]] for the current configuration, which can be used to determine if two
    * identifiers are equal.

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/CatalystTypeConvertersSuite.scala

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst
 
-import java.time.Instant
+import java.time.{Instant, LocalDate}
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.Row
@@ -181,4 +181,39 @@ class CatalystTypeConvertersSuite extends SparkFunSuite with SQLHelper {
       }
     }
   }
+
+  test("converting java.time.LocalDate to DateType") {
+    Seq(
+      "0101-02-16",
+      "1582-10-02",
+      "1582-12-31",
+      "1970-01-01",
+      "1972-12-31",
+      "2019-02-16",
+      "2119-03-16").foreach { timestamp =>
+      val input = LocalDate.parse(timestamp)
+      val result = CatalystTypeConverters.convertToCatalyst(input)
+      val expected = DateTimeUtils.localDateToDays(input)
+      assert(result === expected)
+    }
+  }
+
+  test("converting DateType to java.time.LocalDate") {
+    withSQLConf(SQLConf.DATE_EXTERNAL_TYPE.key -> "LocalDate") {
+      Seq(
+        -701265,
+        -371419,
+        -199722,
+        -1,
+        0,
+        967,
+        2094,
+        17877,
+        24837,
+        1110657).foreach { days =>
+        val localDate = DateTimeUtils.daysToLocalDate(days)
+        assert(CatalystTypeConverters.createToScalaConverter(DateType)(days) === localDate)
+      }
+    }
+  }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala

@@ -294,6 +294,18 @@ class RowEncoderSuite extends CodegenInterpretedPlanTest {
     }
   }
 
+  test("encoding/decoding DateType to/from java.time.LocalDate") {
+    withSQLConf(SQLConf.DATE_EXTERNAL_TYPE.key -> "LocalDate") {
+      val schema = new StructType().add("d", DateType)
+      val encoder = RowEncoder(schema).resolveAndBind()
+      val localDate = java.time.LocalDate.parse("2019-02-27")
+      val row = encoder.toRow(Row(localDate))
+      assert(row.getLong(0) === DateTimeUtils.localDateToDays(localDate))
+      val readback = encoder.fromRow(row)
+      assert(readback.get(0).equals(localDate))
+    }
+  }
+
   for {
     elementType <- Seq(IntegerType, StringType)
     containsNull <- Seq(true, false)

sql/core/src/test/java/test/org/apache/spark/sql/JavaUDFSuite.java

@@ -18,8 +18,10 @@
 package test.org.apache.spark.sql;
 
 import java.io.Serializable;
+import java.time.LocalDate;
 import java.util.List;
 
+import org.apache.spark.sql.internal.SQLConf;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -121,4 +123,20 @@ public void udf6Test() {
     Row result = spark.sql("SELECT returnOne()").head();
     Assert.assertEquals(1, result.getInt(0));
   }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void udf7Test() {
+    String originConf = spark.conf().get(SQLConf.DATE_EXTERNAL_TYPE().key());
+    try {
+      spark.conf().set(SQLConf.DATE_EXTERNAL_TYPE().key(), "LocalDate");
+      spark.udf().register(
+          "plusDay",
+          (java.time.LocalDate ld) -> ld.plusDays(1), DataTypes.DateType);
+      Row result = spark.sql("SELECT plusDay(DATE '2019-02-26')").head();
+      Assert.assertEquals(LocalDate.parse("2019-02-27"), result.get(0));
+    } finally {
+      spark.conf().set(SQLConf.DATE_EXTERNAL_TYPE().key(), originConf);
+    }
+  }
 }

sql/core/src/test/scala/org/apache/spark/sql/UDFSuite.scala

@@ -504,4 +504,14 @@ class UDFSuite extends QueryTest with SharedSQLContext {
       assert(df.collect().toSeq === Seq(Row(expected)))
     }
   }
+
+  test("Using java.time.LocalDate in UDF") {
+    withSQLConf(SQLConf.DATE_EXTERNAL_TYPE.key -> "LocalDate") {
+      val expected = java.time.LocalDate.parse("2019-02-27")
+      val plusDay = udf((i: java.time.LocalDate) => i.plusDays(1))
+      val df = spark.sql("SELECT DATE '2019-02-26' as d")
+        .select(plusDay('d))
+      assert(df.collect().toSeq === Seq(Row(expected)))
+    }
+  }
 }