Support localdate convert to date in datetype (#2780)

core/src/test/scala/com/pingcap/tispark/datatype/BatchWriteDataTypeSuite.scala

@@ -19,12 +19,30 @@ package com.pingcap.tispark.datatype
 import java.sql.{Date, Timestamp}
 import java.util.Calendar
 import com.pingcap.tispark.datasource.BaseBatchWriteTest
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{AnalysisException, Row}
 import org.apache.spark.sql.types._
+import org.scalatest.Matchers.{be, noException, the}
 import org.tikv.common.exception.TiBatchWriteException
 
 class BatchWriteDataTypeSuite extends BaseBatchWriteTest("test_data_type", "test") {
 
+  test("Test date type pr 2780") {
+    val dbtable1 = "test.test_date1"
+    val dbtable2 = "test.test_date2"
+    jdbcUpdate(s"drop table if exists $dbtable1")
+    jdbcUpdate(s"drop table if exists $dbtable2")
+    jdbcUpdate(s"create table $dbtable1 (dt date)")
+    jdbcUpdate(s"create table $dbtable2 (dt date)")
+    jdbcUpdate(s"insert into $dbtable2 values ('2020-01-01')")
+
+    noException should be thrownBy spark
+      .sql(s"insert into $dbtable1 select * from $dbtable2")
+      .show()
+    the[AnalysisException] thrownBy spark
+      .sql(s"insert into $dbtable1 values ('2020-01-01')")
+      .show()
+  }
+
   test("Test Read different types") {
     jdbcUpdate(s"""
                   |create table $dbtable(

tikv-client/src/main/java/com/pingcap/tikv/types/AbstractDateTimeType.java

@@ -25,6 +25,7 @@
 import com.pingcap.tikv.codec.ExtendedDateTime;
 import com.pingcap.tikv.meta.TiColumnInfo.InternalTypeHolder;
 import java.sql.Timestamp;
+import java.time.Instant;
 import org.joda.time.DateTimeZone;
 import org.joda.time.LocalDate;
 import org.tikv.common.exception.ConvertNotSupportException;
@@ -130,6 +131,8 @@ java.sql.Timestamp convertToMysqlDateTime(Object value) throws ConvertNotSupport
       result = new java.sql.Timestamp(((java.sql.Date) value).getTime());
     } else if (value instanceof java.sql.Timestamp) {
       result = (java.sql.Timestamp) value;
+    } else if (value instanceof Instant) {
+      result = java.sql.Timestamp.from((Instant) value);
     } else {
       throw new ConvertNotSupportException(value.getClass().getName(), this.getClass().getName());
     }

tikv-client/src/main/java/com/pingcap/tikv/types/DateType.java

@@ -68,6 +68,8 @@ private java.sql.Date convertToMysqlDate(Object value) throws ConvertNotSupportE
       result = (java.sql.Date) value;
     } else if (value instanceof java.sql.Timestamp) {
       result = new java.sql.Date(((java.sql.Timestamp) value).getTime());
+    } else if (value instanceof java.time.LocalDate) {
+      result = java.sql.Date.valueOf(((java.time.LocalDate) value).toString());
     } else {
       throw new ConvertNotSupportException(value.getClass().getName(), this.getClass().getName());
     }