[SPARK-29387][SQL] Support * and / operators for intervals

common/unsafe/src/main/java/org/apache/spark/unsafe/types/CalendarInterval.java

@@ -59,6 +59,27 @@ public CalendarInterval subtract(CalendarInterval that) {
     return new CalendarInterval(months, days, microseconds);
   }
 
+  private static CalendarInterval fromDoubles(
+      double monthsWithFraction,
+      double daysWithFraction,
+      double microsWithFraction) {
+    int truncatedMonths = Math.toIntExact((long)(monthsWithFraction));
+    // Using 30 days per month as PostgreSQL does.
+    double days = daysWithFraction + 30 * (monthsWithFraction - truncatedMonths);
+    int truncatedDays = Math.toIntExact((long)days);
+    double micros = microsWithFraction + MICROS_PER_DAY * (days - truncatedDays);
+    long truncatedMicros = Math.round(micros);
+    return new CalendarInterval(truncatedMonths, truncatedDays, truncatedMicros);
+  }
+
+  public CalendarInterval multiply(double num) {
+    return fromDoubles(num * this.months, num * this.days, num * this.microseconds);
+  }
+
+  public CalendarInterval divide(double num) {
+    return fromDoubles(this.months / num, this.days / num, this.microseconds / num);
+  }
+
   public CalendarInterval negate() {
     return new CalendarInterval(-this.months, -this.days, -this.microseconds);
   }

common/unsafe/src/test/java/org/apache/spark/unsafe/types/CalendarIntervalSuite.java

@@ -94,4 +94,61 @@ public void subtractTest() {
     input2 = new CalendarInterval(75, 150, 200 * MICROS_PER_HOUR);
     assertEquals(input1.subtract(input2), new CalendarInterval(-85, -180, -281 * MICROS_PER_HOUR));
   }
+
+  @Test
+  public void multiplyTest() {
+    CalendarInterval interval = new CalendarInterval(0, 0, 0);
+    assertEquals(interval, interval.multiply(0));
+
+    interval = new CalendarInterval(123, 456, 789);
+    assertEquals(new CalendarInterval(123 * 42, 456 * 42, 789 * 42), interval.multiply(42));
+
+    interval = new CalendarInterval(-123, -456, -789);
+    assertEquals(new CalendarInterval(-123 * 42, -456 * 42, -789 * 42), interval.multiply(42));
+
+    assertEquals(
+      new CalendarInterval(1, 22, 12 * MICROS_PER_HOUR),
+      new CalendarInterval(1, 5, 0).multiply(1.5));
+    assertEquals(
+      new CalendarInterval(2, 14, 12 * MICROS_PER_HOUR),
+      new CalendarInterval(2, 2, 2 * MICROS_PER_HOUR).multiply(1.2));
+
+    try {
+      interval = new CalendarInterval(2, 0, 0);
+      interval.multiply(Integer.MAX_VALUE);
+      fail("Expected to throw an exception on months overflow");
+    } catch (java.lang.ArithmeticException e) {
+      assertTrue(e.getMessage().contains("overflow"));
+    }
+  }
+
+    @Test
+    public void divideTest() {
+      CalendarInterval interval = new CalendarInterval(0, 0, 0);
+      assertEquals(interval, interval.divide(10));
+
+      interval = new CalendarInterval(1, 3, 30 * MICROS_PER_SECOND);
+      assertEquals(
+        new CalendarInterval(0, 16, 12 * MICROS_PER_HOUR + 15 * MICROS_PER_SECOND),
+        interval.divide(2));
+      assertEquals(
+        new CalendarInterval(2, 6, MICROS_PER_MINUTE),
+        interval.divide(0.5));
+
+      interval = new CalendarInterval(-1, 0, -30 * MICROS_PER_SECOND);
+      assertEquals(
+        new CalendarInterval(0, -15, -15 * MICROS_PER_SECOND),
+        interval.divide(2));
+      assertEquals(
+        new CalendarInterval(-2, 0, -1 * MICROS_PER_MINUTE),
+        interval.divide(0.5));
+
+      try {
+        interval = new CalendarInterval(123, 456, 789);
+        interval.divide(0);
+        fail("Expected to throw an exception on divide by zero");
+      } catch (java.lang.ArithmeticException e) {
+        assertTrue(e.getMessage().contains("overflow"));
+      }
+    }
 }

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

@@ -831,6 +831,8 @@ object TypeCoercion {
    * 2. Turns Add/Subtract of TimestampType/DateType/IntegerType
    *    and TimestampType/IntegerType/DateType to DateAdd/DateSub/SubtractDates and
    *    to SubtractTimestamps.
+   * 3. Turns Multiply/Divide of CalendarIntervalType and NumericType
+   *    to MultiplyInterval/DivideInterval
    */
   object DateTimeOperations extends Rule[LogicalPlan] {
 
@@ -846,6 +848,12 @@ object TypeCoercion {
         Cast(TimeAdd(l, r), l.dataType)
       case Subtract(l, r @ CalendarIntervalType()) if acceptedTypes.contains(l.dataType) =>
         Cast(TimeSub(l, r), l.dataType)
+      case Multiply(l @ CalendarIntervalType(), r @ NumericType()) =>
+        MultiplyInterval(l, r)
+      case Multiply(l @ NumericType(), r @ CalendarIntervalType()) =>
+        MultiplyInterval(r, l)
+      case Divide(l @ CalendarIntervalType(), r @ NumericType()) =>
+        DivideInterval(l, r)
 
       case Add(l @ DateType(), r @ IntegerType()) => DateAdd(l, r)
       case Add(l @ IntegerType(), r @ DateType()) => DateAdd(r, l)

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

@@ -45,45 +45,45 @@ abstract class ExtractIntervalPart(
 }
 
 case class ExtractIntervalMillenniums(child: Expression)
-    extends ExtractIntervalPart(child, IntegerType, getMillenniums, "getMillenniums")
+  extends ExtractIntervalPart(child, IntegerType, getMillenniums, "getMillenniums")
 
 case class ExtractIntervalCenturies(child: Expression)
-    extends ExtractIntervalPart(child, IntegerType, getCenturies, "getCenturies")
+  extends ExtractIntervalPart(child, IntegerType, getCenturies, "getCenturies")
 
 case class ExtractIntervalDecades(child: Expression)
-    extends ExtractIntervalPart(child, IntegerType, getDecades, "getDecades")
+  extends ExtractIntervalPart(child, IntegerType, getDecades, "getDecades")
 
 case class ExtractIntervalYears(child: Expression)
-    extends ExtractIntervalPart(child, IntegerType, getYears, "getYears")
+  extends ExtractIntervalPart(child, IntegerType, getYears, "getYears")
 
 case class ExtractIntervalQuarters(child: Expression)
-    extends ExtractIntervalPart(child, ByteType, getQuarters, "getQuarters")
+  extends ExtractIntervalPart(child, ByteType, getQuarters, "getQuarters")
 
 case class ExtractIntervalMonths(child: Expression)
-    extends ExtractIntervalPart(child, ByteType, getMonths, "getMonths")
+  extends ExtractIntervalPart(child, ByteType, getMonths, "getMonths")
 
 case class ExtractIntervalDays(child: Expression)
-    extends ExtractIntervalPart(child, IntegerType, getDays, "getDays")
+  extends ExtractIntervalPart(child, IntegerType, getDays, "getDays")
 
 case class ExtractIntervalHours(child: Expression)
-    extends ExtractIntervalPart(child, LongType, getHours, "getHours")
+  extends ExtractIntervalPart(child, LongType, getHours, "getHours")
 
 case class ExtractIntervalMinutes(child: Expression)
-    extends ExtractIntervalPart(child, ByteType, getMinutes, "getMinutes")
+  extends ExtractIntervalPart(child, ByteType, getMinutes, "getMinutes")
 
 case class ExtractIntervalSeconds(child: Expression)
-    extends ExtractIntervalPart(child, DecimalType(8, 6), getSeconds, "getSeconds")
+  extends ExtractIntervalPart(child, DecimalType(8, 6), getSeconds, "getSeconds")
 
 case class ExtractIntervalMilliseconds(child: Expression)
-    extends ExtractIntervalPart(child, DecimalType(8, 3), getMilliseconds, "getMilliseconds")
+  extends ExtractIntervalPart(child, DecimalType(8, 3), getMilliseconds, "getMilliseconds")
 
 case class ExtractIntervalMicroseconds(child: Expression)
-    extends ExtractIntervalPart(child, LongType, getMicroseconds, "getMicroseconds")
+  extends ExtractIntervalPart(child, LongType, getMicroseconds, "getMicroseconds")
 
 // Number of seconds in 10000 years is 315576000001 (30 days per one month)
 // which is 12 digits + 6 digits for the fractional part of seconds.
 case class ExtractIntervalEpoch(child: Expression)
-    extends ExtractIntervalPart(child, DecimalType(18, 6), getEpoch, "getEpoch")
+  extends ExtractIntervalPart(child, DecimalType(18, 6), getEpoch, "getEpoch")
 
 object ExtractIntervalPart {
 
@@ -109,3 +109,54 @@ object ExtractIntervalPart {
     case _ => errorHandleFunc
   }
 }
+
+abstract class IntervalNumOperation(
+    interval: Expression,
+    num: Expression,
+    operation: (CalendarInterval, Double) => CalendarInterval,
+    operationName: String)
+  extends BinaryExpression with ImplicitCastInputTypes with Serializable {
+  override def left: Expression = interval
+  override def right: Expression = num
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(CalendarIntervalType, DoubleType)
+  override def dataType: DataType = CalendarIntervalType
+
+  override def nullable: Boolean = true
+
+  override def nullSafeEval(interval: Any, num: Any): Any = {
+    try {
+      operation(interval.asInstanceOf[CalendarInterval], num.asInstanceOf[Double])
+    } catch {
+      case _: java.lang.ArithmeticException => null
+    }
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    nullSafeCodeGen(ctx, ev, (interval, num) => {
+      s"""
+        try {
+          ${ev.value} = $interval.$operationName($num);
+        } catch (java.lang.ArithmeticException e) {
+          ${ev.isNull} = true;
+        }
+      """
+    })
+  }
+
+  override def prettyName: String = operationName + "_interval"
+}
+
+case class MultiplyInterval(interval: Expression, num: Expression)
+  extends IntervalNumOperation(
+    interval,
+    num,
+    (i: CalendarInterval, n: Double) => i.multiply(n),
+    "multiply")
+
+case class DivideInterval(interval: Expression, num: Expression)
+  extends IntervalNumOperation(
+    interval,
+    num,
+    (i: CalendarInterval, n: Double) => i.divide(n),
+    "divide")

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

@@ -1597,6 +1597,27 @@ class TypeCoercionSuite extends AnalysisTest {
       Multiply(CaseWhen(Seq((EqualTo(1, 2), Cast(1, DecimalType(34, 24)))),
         Cast(100, DecimalType(34, 24))), Cast(1, IntegerType)))
   }
+
+  test("rule for interval operations") {
+    val dateTimeOperations = TypeCoercion.DateTimeOperations
+    val interval = Literal(new CalendarInterval(0, 0, 0))
+
+    Seq(
+      Literal(10.toByte, ByteType),
+      Literal(10.toShort, ShortType),
+      Literal(10, IntegerType),
+      Literal(10L, LongType),
+      Literal(Decimal(10), DecimalType.SYSTEM_DEFAULT),
+      Literal(10.5.toFloat, FloatType),
+      Literal(10.5, DoubleType)).foreach { num =>
+      ruleTest(dateTimeOperations, Multiply(interval, num),
+        MultiplyInterval(interval, num))
+      ruleTest(dateTimeOperations, Multiply(num, interval),
+        MultiplyInterval(interval, num))
+      ruleTest(dateTimeOperations, Divide(interval, num),
+        DivideInterval(interval, num))
+    }
+  }
 }
 
 

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

@@ -20,12 +20,12 @@ package org.apache.spark.sql.catalyst.expressions
 import scala.language.implicitConversions
 
 import org.apache.spark.SparkFunSuite
-import org.apache.spark.sql.catalyst.util.IntervalUtils
+import org.apache.spark.sql.catalyst.util.IntervalUtils.fromString
 import org.apache.spark.sql.types.Decimal
 
 class IntervalExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
   implicit def interval(s: String): Literal = {
-    Literal(IntervalUtils.fromString("interval " + s))
+    Literal(fromString("interval " + s))
   }
 
   test("millenniums") {
@@ -191,4 +191,52 @@ class IntervalExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
       ExtractIntervalEpoch("1 second 1 millisecond 1 microsecond"),
       Decimal(1.001001, 18, 6))
   }
+
+  test("multiply") {
+    def multiply(interval: String, num: Double): Expression = {
+      MultiplyInterval(Literal(fromString(interval)), Literal(num))
+    }
+    checkEvaluation(multiply("0 seconds", 10), fromString("0 seconds"))
+    checkEvaluation(multiply("10 hours", 0), fromString("0 hours"))
+    checkEvaluation(multiply("12 months 1 microseconds", 2), fromString("2 years 2 microseconds"))
+    checkEvaluation(multiply("-5 year 3 seconds", 3), fromString("-15 years 9 seconds"))
+    checkEvaluation(multiply("1 year 1 second", 0.5), fromString("6 months 500 milliseconds"))
+    checkEvaluation(
+      multiply("-100 years -1 millisecond", 0.5),
+      fromString("-50 years -500 microseconds"))
+    checkEvaluation(
+      multiply("2 months 4 seconds", -0.5),
+      fromString("-1 months -2 seconds"))
+    checkEvaluation(
+      multiply("1 month 2 microseconds", 1.5),
+      fromString("1 months 15 days 3 microseconds"))
+    checkEvaluation(multiply("2 months", Int.MaxValue), null)
+  }
+
+  test("divide") {
+    def divide(interval: String, num: Double): Expression = {
+      DivideInterval(Literal(fromString(interval)), Literal(num))
+    }
+
+    checkEvaluation(divide("0 seconds", 10), fromString("0 seconds"))
+    checkEvaluation(
+      divide("12 months 3 milliseconds", 2),
+      fromString("6 months 1 milliseconds 500 microseconds"))
+    checkEvaluation(
+      divide("-5 year 3 seconds", 3),
+      fromString("-1 years -8 months 1 seconds"))
+    checkEvaluation(
+      divide("6 years -7 seconds", 3),
+      fromString("2 years -2 seconds -333 milliseconds -333 microseconds"))
+    checkEvaluation(
+      divide("2 years -8 seconds", 0.5),
+      fromString("4 years -16 seconds"))
+    checkEvaluation(
+      divide("-1 month 2 microseconds", -0.25),
+      fromString("4 months -8 microseconds"))
+    checkEvaluation(
+      divide("1 month 3 microsecond", 1.5),
+      fromString("20 days 2 microseconds"))
+    checkEvaluation(divide("2 months", 0), null)
+  }
 }

sql/core/src/test/resources/sql-tests/inputs/datetime.sql

@@ -36,3 +36,8 @@ select date '2001-10-01' - 7;
 select date '2001-10-01' - date '2001-09-28';
 select date'2020-01-01' - timestamp'2019-10-06 10:11:12.345678';
 select timestamp'2019-10-06 10:11:12.345678' - date'2020-01-01';
+
+-- interval operations
+select 3 * (timestamp'2019-10-15 10:11:12.001002' - date'2019-10-15');
+select interval 4 month 2 weeks 3 microseconds * 1.5;
+select (timestamp'2019-10-15' - timestamp'2019-10-14') / 1.5;

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

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 17
+-- Number of queries: 20
 
 
 -- !query 0
@@ -145,3 +145,27 @@ select timestamp'2019-10-06 10:11:12.345678' - date'2020-01-01'
 struct<subtracttimestamps(TIMESTAMP('2019-10-06 10:11:12.345678'), CAST(DATE '2020-01-01' AS TIMESTAMP)):interval>
 -- !query 16 output
 interval -2078 hours -48 minutes -47 seconds -654 milliseconds -322 microseconds
+
+
+-- !query 17
+select 3 * (timestamp'2019-10-15 10:11:12.001002' - date'2019-10-15')
+-- !query 17 schema
+struct<multiply_interval(subtracttimestamps(TIMESTAMP('2019-10-15 10:11:12.001002'), CAST(DATE '2019-10-15' AS TIMESTAMP)), CAST(3 AS DOUBLE)):interval>
+-- !query 17 output
+interval 30 hours 33 minutes 36 seconds 3 milliseconds 6 microseconds
+
+
+-- !query 18
+select interval 4 month 2 weeks 3 microseconds * 1.5
+-- !query 18 schema
+struct<multiply_interval(interval 4 months 2 weeks 3 microseconds, CAST(1.5 AS DOUBLE)):interval>
+-- !query 18 output
+interval 6 months 3 weeks 5 microseconds
+
+
+-- !query 19
+select (timestamp'2019-10-15' - timestamp'2019-10-14') / 1.5
+-- !query 19 schema
+struct<divide_interval(subtracttimestamps(TIMESTAMP('2019-10-15 00:00:00'), TIMESTAMP('2019-10-14 00:00:00')), CAST(1.5 AS DOUBLE)):interval>
+-- !query 19 output
+interval 16 hours