[SPARK-29364][SQL] Return an interval from date subtract according to SQL standard

docs/sql-migration-guide.md

@@ -217,6 +217,8 @@ license: |
 
   - Since Spark 3.0, the `size` function returns `NULL` for the `NULL` input. In Spark version 2.4 and earlier, this function gives `-1` for the same input. To restore the behavior before Spark 3.0, you can set `spark.sql.legacy.sizeOfNull` to `true`.
 
+  - In Spark version 2.4 and earlier, dates subtraction `date1` - `date2` gives the number of days from `date1` to `date2`. Since Spark 3.0, the expression has the `INTERVAL` type and returns an interval between two dates. To get the number of days, use the `datediff` function.
+
 ## Upgrading from Spark SQL 2.4 to 2.4.1
 
   - The value of `spark.executor.heartbeatInterval`, when specified without units like "30" rather than "30s", was

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

@@ -828,8 +828,8 @@ object TypeCoercion {
   /**
    * 1. Turns Add/Subtract of DateType/TimestampType/StringType and CalendarIntervalType
    *    to TimeAdd/TimeSub.
-   * 2. Turns Add/Subtract of DateType/IntegerType and IntegerType/DateType
-   *    to DateAdd/DateSub/DateDiff.
+   * 2. Turns Add/Subtract of TimestampType/DateType/IntegerType
+   *    and TimestampType/IntegerType/DateType to DateAdd/DateSub/SubtractDates.
    */
   object DateTimeOperations extends Rule[LogicalPlan] {
 
@@ -849,7 +849,7 @@ object TypeCoercion {
       case Add(l @ DateType(), r @ IntegerType()) => DateAdd(l, r)
       case Add(l @ IntegerType(), r @ DateType()) => DateAdd(r, l)
       case Subtract(l @ DateType(), r @ IntegerType()) => DateSub(l, r)
-      case Subtract(l @ DateType(), r @ DateType()) => DateDiff(l, r)
+      case Subtract(l @ DateType(), r @ DateType()) => SubtractDates(l, r)
       case Subtract(l @ TimestampType(), r @ TimestampType()) => TimestampDiff(l, r)
       case Subtract(l @ TimestampType(), r @ DateType()) =>
         TimestampDiff(l, Cast(r, TimestampType))

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala

@@ -2116,3 +2116,25 @@ case class TimestampDiff(endTimestamp: Expression, startTimestamp: Expression)
       s"new org.apache.spark.unsafe.types.CalendarInterval(0, $end - $start)")
   }
 }
+
+/**
+ * Returns the interval from the `left` date (inclusive) to the `right` date (exclusive).
+ */
+case class SubtractDates(left: Expression, right: Expression)
+  extends BinaryExpression with ImplicitCastInputTypes {
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(DateType, DateType)
+  override def dataType: DataType = CalendarIntervalType
+
+  override def nullSafeEval(leftDays: Any, rightDays: Any): Any = {
+    DateTimeUtils.subtractDates(leftDays.asInstanceOf[Int], rightDays.asInstanceOf[Int])
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    defineCodeGen(ctx, ev, (leftDays, rightDays) => {
+      val dtu = DateTimeUtils.getClass.getName.stripSuffix("$")
+      s"$dtu.subtractDates($leftDays, $rightDays)"
+    })
+  }
+}
+

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

@@ -27,7 +27,7 @@ import java.util.concurrent.TimeUnit._
 import scala.util.control.NonFatal
 
 import org.apache.spark.sql.types.Decimal
-import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
 
 /**
  * Helper functions for converting between internal and external date and time representations.
@@ -950,4 +950,20 @@ object DateTimeUtils {
       None
     }
   }
+
+  /**
+   * Subtracts two dates.
+   * @param endDate - the end date, exclusive
+   * @param startDate - the start date, inclusive
+   * @return an interval between two dates. The interval can be negative
+   *         if the end date is before the start date.
+   */
+  def subtractDates(endDate: SQLDate, startDate: SQLDate): CalendarInterval = {
+    val period = Period.between(
+      LocalDate.ofEpochDay(startDate),
+      LocalDate.ofEpochDay(endDate))
+    val months = period.getMonths + 12 * period.getYears
+    val microseconds = period.getDays * MICROS_PER_DAY
+    new CalendarInterval(months, microseconds)
+  }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercionSuite.scala

@@ -1430,7 +1430,7 @@ class TypeCoercionSuite extends AnalysisTest {
     ruleTest(dateTimeOperations, Add(date, intValue), DateAdd(date, intValue))
     ruleTest(dateTimeOperations, Add(intValue, date), DateAdd(date, intValue))
     ruleTest(dateTimeOperations, Subtract(date, intValue), DateSub(date, intValue))
-    ruleTest(dateTimeOperations, Subtract(date, date), DateDiff(date, date))
+    ruleTest(dateTimeOperations, Subtract(date, date), SubtractDates(date, date))
     ruleTest(dateTimeOperations, Subtract(timestamp, timestamp),
       TimestampDiff(timestamp, timestamp))
     ruleTest(dateTimeOperations, Subtract(timestamp, date),

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.catalyst.expressions
 
 import java.sql.{Date, Timestamp}
 import java.text.SimpleDateFormat
-import java.time.{Instant, LocalDateTime, ZoneId, ZoneOffset}
+import java.time.{Instant, LocalDate, LocalDateTime, ZoneId, ZoneOffset}
 import java.util.{Calendar, Locale, TimeZone}
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.TimeUnit._
@@ -1087,4 +1087,24 @@ class DateExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
       CalendarInterval.fromString("interval 521722 weeks 4 days " +
         "23 hours 59 minutes 59 seconds 999 milliseconds 999 microseconds"))
   }
+
+  test("subtract dates") {
+    val end = LocalDate.of(2019, 10, 5)
+    checkEvaluation(SubtractDates(Literal(end), Literal(end)),
+      new CalendarInterval(0, 0))
+    checkEvaluation(SubtractDates(Literal(end.plusDays(1)), Literal(end)),
+      CalendarInterval.fromString("interval 1 days"))
+    checkEvaluation(SubtractDates(Literal(end.minusDays(1)), Literal(end)),
+      CalendarInterval.fromString("interval -1 days"))
+    val epochDate = Literal(LocalDate.ofEpochDay(0))
+    checkEvaluation(SubtractDates(Literal(end), epochDate),
+      CalendarInterval.fromString("interval 49 years 9 months 4 days"))
+    checkEvaluation(SubtractDates(epochDate, Literal(end)),
+      CalendarInterval.fromString("interval -49 years -9 months -4 days"))
+    checkEvaluation(
+      SubtractDates(
+        Literal(LocalDate.of(10000, 1, 1)),
+        Literal(LocalDate.of(1, 1, 1))),
+      CalendarInterval.fromString("interval 9999 years"))
+  }
 }

sql/core/src/test/resources/sql-tests/results/datetime.sql.out

@@ -126,9 +126,9 @@ struct<date_sub(DATE '2001-10-01', 7):date>
 -- !query 14
 select date '2001-10-01' - date '2001-09-28'
 -- !query 14 schema
-struct<datediff(DATE '2001-10-01', DATE '2001-09-28'):int>
+struct<subtractdates(DATE '2001-10-01', DATE '2001-09-28'):interval>
 -- !query 14 output
-3
+interval 3 days
 
 
 -- !query 15

sql/core/src/test/resources/sql-tests/results/postgreSQL/date.sql.out

@@ -482,93 +482,93 @@ SELECT date '5874898-01-01'
 -- !query 46
 SELECT f1 - date '2000-01-01' AS `Days From 2K` FROM DATE_TBL
 -- !query 46 schema
-struct<Days From 2K:int>
+struct<Days From 2K:interval>
 -- !query 46 output
--1035
--1036
--1037
--1400
--1401
--1402
--1403
--15542
--15607
-13977
-14343
-14710
-91
-92
-93
+interval -2 years -10 months
+interval -2 years -10 months -1 days
+interval -2 years -9 months -4 weeks -2 days
+interval -3 years -10 months
+interval -3 years -10 months -1 days
+interval -3 years -10 months -2 days
+interval -3 years -9 months -4 weeks -2 days
+interval -42 years -6 months -2 weeks -4 days
+interval -42 years -8 months -3 weeks -1 days
+interval 3 months
+interval 3 months 1 days
+interval 3 months 2 days
+interval 38 years 3 months 1 weeks
+interval 39 years 3 months 1 weeks 1 days
+interval 40 years 3 months 1 weeks 2 days
 
 
 -- !query 47
 SELECT f1 - date 'epoch' AS `Days From Epoch` FROM DATE_TBL
 -- !query 47 schema
-struct<Days From Epoch:int>
+struct<Days From Epoch:interval>
 -- !query 47 output
--4585
--4650
-11048
-11049
-11050
-24934
-25300
-25667
-9554
-9555
-9556
-9557
-9920
-9921
-9922
+interval -12 years -6 months -2 weeks -4 days
+interval -12 years -8 months -3 weeks -1 days
+interval 26 years 1 months 3 weeks 6 days
+interval 26 years 1 months 4 weeks
+interval 26 years 2 months
+interval 26 years 2 months 1 days
+interval 27 years 1 months 3 weeks 6 days
+interval 27 years 2 months
+interval 27 years 2 months 1 days
+interval 30 years 3 months
+interval 30 years 3 months 1 days
+interval 30 years 3 months 2 days
+interval 68 years 3 months 1 weeks
+interval 69 years 3 months 1 weeks 1 days
+interval 70 years 3 months 1 weeks 2 days
 
 
 -- !query 48
 SELECT date 'yesterday' - date 'today' AS `One day`
 -- !query 48 schema
-struct<One day:int>
+struct<One day:interval>
 -- !query 48 output
--1
+interval -1 days
 
 
 -- !query 49
 SELECT date 'today' - date 'tomorrow' AS `One day`
 -- !query 49 schema
-struct<One day:int>
+struct<One day:interval>
 -- !query 49 output
--1
+interval -1 days
 
 
 -- !query 50
 SELECT date 'yesterday' - date 'tomorrow' AS `Two days`
 -- !query 50 schema
-struct<Two days:int>
+struct<Two days:interval>
 -- !query 50 output
--2
+interval -2 days
 
 
 -- !query 51
 SELECT date 'tomorrow' - date 'today' AS `One day`
 -- !query 51 schema
-struct<One day:int>
+struct<One day:interval>
 -- !query 51 output
-1
+interval 1 days
 
 
 -- !query 52
 SELECT date 'today' - date 'yesterday' AS `One day`
 -- !query 52 schema
-struct<One day:int>
+struct<One day:interval>
 -- !query 52 output
-1
+interval 1 days
 
 
 -- !query 53
 SELECT date 'tomorrow' - date 'yesterday' AS `Two days`
 -- !query 53 schema
-struct<Two days:int>
+struct<Two days:interval>
 -- !query 53 output
-2
+interval 2 days
 
 
 -- !query 54