Negate intervals

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

@@ -83,12 +83,19 @@ case class UnaryMinus(
       val iu = IntervalUtils.getClass.getCanonicalName.stripSuffix("$")
       val method = if (failOnError) "negateExact" else "negate"
       defineCodeGen(ctx, ev, c => s"$iu.$method($c)")
+    case DayTimeIntervalType | YearMonthIntervalType =>
+      nullSafeCodeGen(ctx, ev, eval => {
+        val mathClass = classOf[Math].getName
+        s"${ev.value} = $mathClass.negateExact($eval);"
+      })
   }
 
   protected override def nullSafeEval(input: Any): Any = dataType match {
     case CalendarIntervalType if failOnError =>
       IntervalUtils.negateExact(input.asInstanceOf[CalendarInterval])
     case CalendarIntervalType => IntervalUtils.negate(input.asInstanceOf[CalendarInterval])
+    case DayTimeIntervalType => Math.negateExact(input.asInstanceOf[Long])
+    case YearMonthIntervalType => Math.negateExact(input.asInstanceOf[Int])
     case _ => numeric.negate(input)
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/types/AbstractDataType.scala

@@ -82,7 +82,11 @@ private[sql] object TypeCollection {
    * Types that include numeric types and interval type. They are only used in unary_minus,
    * unary_positive, add and subtract operations.
    */
-  val NumericAndInterval = TypeCollection(NumericType, CalendarIntervalType)
+  val NumericAndInterval = TypeCollection(
+    NumericType,
+    CalendarIntervalType,
+    DayTimeIntervalType,
+    YearMonthIntervalType)
 
   def apply(types: AbstractDataType*): TypeCollection = new TypeCollection(types)
 

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions
 
 import java.sql.{Date, Timestamp}
+import java.time.Period
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.sql.catalyst.InternalRow
@@ -102,6 +103,9 @@ class ArithmeticExpressionSuite extends SparkFunSuite with ExpressionEvalHelper
     checkEvaluation(UnaryMinus(negativeIntLit), - negativeInt)
     checkEvaluation(UnaryMinus(positiveLongLit), - positiveLong)
     checkEvaluation(UnaryMinus(negativeLongLit), - negativeLong)
+    checkExceptionInExpression[ArithmeticException](
+      UnaryMinus(Literal.create(Period.ofMonths(Int.MinValue), YearMonthIntervalType)),
+      "overflow")
 
     Seq("true", "false").foreach { failOnError =>
       withSQLConf(SQLConf.ANSI_ENABLED.key -> failOnError) {

sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeTestUtils.scala

@@ -52,7 +52,10 @@ object DataTypeTestUtils {
   /**
    * Instances of all [[NumericType]]s and [[CalendarIntervalType]]
    */
-  val numericAndInterval: Set[DataType] = numericTypeWithoutDecimal + CalendarIntervalType
+  val numericAndInterval: Set[DataType] = numericTypeWithoutDecimal ++ Set(
+    CalendarIntervalType,
+    DayTimeIntervalType,
+    YearMonthIntervalType)
 
   /**
    * All the types that support ordering

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

@@ -18,6 +18,7 @@
 package org.apache.spark.sql
 
 import java.sql.{Date, Timestamp}
+import java.time.{Duration, Period}
 import java.util.Locale
 
 import org.apache.hadoop.io.{LongWritable, Text}
@@ -2375,4 +2376,12 @@ class ColumnExpressionSuite extends QueryTest with SharedSparkSession {
     assert(e2.getCause.isInstanceOf[RuntimeException])
     assert(e2.getCause.getMessage == "hello")
   }
+
+  test("negate year-month and day-time intervals") {
+    import testImplicits._
+    val df = Seq((Period.ofMonths(10), Duration.ofDays(10)))
+      .toDF("year-month", "day-time")
+    val negated = df.select(-$"year-month", -$"day-time")
+    checkAnswer(negated, Row(Period.ofMonths(-10), Duration.ofDays(-10)))
+  }
 }