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

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

@@ -355,6 +355,22 @@ public CalendarInterval subtract(CalendarInterval that) {
     return new CalendarInterval(months, microseconds);
   }
 
+  private static CalendarInterval fromDoubles(double months, double microseconds) {
+    long roundedMonths = (long)(months);
+    // Using 30 days per month as PostgreSQL does.
+    double microsInMonth = 30 * MICROS_PER_DAY * (months - roundedMonths);
+    long roundedMicroseconds = (long)(microseconds + microsInMonth);
+    return new CalendarInterval(Math.toIntExact(roundedMonths), roundedMicroseconds);
+  }
+
+  public CalendarInterval multiply(double num) {
+    return fromDoubles(num * this.months, num * this.microseconds);
+  }
+
+  public CalendarInterval divide(double num) {
+    return fromDoubles(this.months / num, this.microseconds / num);
+  }
+
   public CalendarInterval negate() {
     return new CalendarInterval(-this.months, -this.microseconds);
   }

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

@@ -297,4 +297,50 @@ public void fromStringCaseSensitivityTest() {
     assertNull(fromString("INTERVAL"));
     assertNull(fromString("  Interval "));
   }
+
+  @Test
+  public void multiplyTest() {
+    CalendarInterval interval = new CalendarInterval(0, 0);
+    assertEquals(interval, interval.multiply(0));
+
+    interval = new CalendarInterval(123, 456);
+    assertEquals(new CalendarInterval(123 * 42, 456 * 42), interval.multiply(42));
+
+    interval = new CalendarInterval(-123, -456);
+    assertEquals(new CalendarInterval(-123 * 42, -456 * 42), interval.multiply(42));
+
+    assertEquals(
+      fromString("interval 1 month 22 days 12 hours"),
+      fromString("interval 1 month 5 days").multiply(1.5));
+
+    try {
+      interval = new CalendarInterval(2, 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);
+      assertEquals(interval, interval.divide(10));
+
+      interval = fromString("1 month 30 seconds");
+      assertEquals(fromString("15 days 15 seconds"), interval.divide(2));
+      assertEquals(fromString("2 months 1 minute"), interval.divide(0.5));
+
+      interval = fromString("-1 month -30 seconds");
+      assertEquals(fromString("-15 days -15 seconds"), interval.divide(2));
+      assertEquals(fromString("-2 months -1 minute"), interval.divide(0.5));
+
+      try {
+        interval = new CalendarInterval(123, 456);
+        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

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext, ExprCode}
+import org.apache.spark.sql.types.{AbstractDataType, CalendarIntervalType, DataType, DoubleType}
+import org.apache.spark.unsafe.types.CalendarInterval
+
+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))
+
+    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

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.unsafe.types.CalendarInterval.fromString
+
+class IntervalExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
+  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 -12 weeks -2 days -14 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 1 days 6 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 4 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