[SPARK-29520][SS] Fix checks of negative intervals

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

@@ -18,6 +18,7 @@
 package org.apache.spark.unsafe.types;
 
 import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
@@ -409,4 +410,25 @@ private void appendUnit(StringBuilder sb, long value, String unit) {
       sb.append(' ').append(value).append(' ').append(unit).append('s');
     }
   }
+
+  /**
+   * Gets interval duration
+   * @param daysPerMonth the number of days per one month
+   * @param targetUnit time units of the result
+   * @return duration in the specified time units
+   */
+  public long getDuration(int daysPerMonth, TimeUnit targetUnit) {
+    long monthsDuration = Math.multiplyExact(daysPerMonth * MICROS_PER_DAY, months);
+    long result = Math.addExact(microseconds, monthsDuration);
+    return targetUnit.convert(result, TimeUnit.MICROSECONDS);
+  }
+
+  /**
+   * Checks the interval is negative
+   * @param daysPerMonth the number of days per one month
+   * @return true if duration of the given interval is less than 0 otherwise false
+   */
+  public boolean isNegative(int daysPerMonth) {
+    return getDuration(daysPerMonth, TimeUnit.MICROSECONDS) < 0;
+  }
 }

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

@@ -20,6 +20,7 @@
 import org.junit.Test;
 
 import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.*;
 import static org.apache.spark.unsafe.types.CalendarInterval.*;
@@ -297,4 +298,32 @@ public void fromStringCaseSensitivityTest() {
     assertNull(fromString("INTERVAL"));
     assertNull(fromString("  Interval "));
   }
+
+  @Test
+  public void durationTest() {
+    assertEquals(fromString("0 seconds").getDuration(31, TimeUnit.MILLISECONDS), 0);
+    assertEquals(fromString("1 month").getDuration(31, TimeUnit.DAYS), 31);
+    assertEquals(fromString("1 microsecond").getDuration(30, TimeUnit.MICROSECONDS), 1);
+    assertEquals(fromString("1 month -30 days").getDuration(31, TimeUnit.DAYS), 1);
+
+    try {
+      fromString(Integer.MAX_VALUE + " month").getDuration(31, TimeUnit.SECONDS);
+      fail("Expected to throw an exception for the invalid input");
+    } catch (ArithmeticException e) {
+      assertTrue(e.getMessage().contains("overflow"));
+    }
+  }
+
+  @Test
+  public void negativeIntervalTest() {
+    assertTrue(fromString("-1 months").isNegative(28));
+    assertTrue(fromString("-1 microsecond").isNegative(30));
+    assertTrue(fromString("-1 month 30 days").isNegative(31));
+    assertTrue(fromString("2 months -61 days").isNegative(30));
+    assertTrue(fromString("-1 year -2 seconds").isNegative(30));
+
+    assertFalse(fromString("0 months").isNegative(28));
+    assertFalse(fromString("1 year -360 days").isNegative(31));
+    assertFalse(fromString("-1 year 380 days").isNegative(31));
+  }
 }

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

@@ -256,7 +256,7 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
           val castedLit = lit.dataType match {
             case CalendarIntervalType =>
               val calendarInterval = lit.value.asInstanceOf[CalendarInterval]
-              if (calendarInterval.months > 0) {
+              if (calendarInterval.months != 0) {
                 invalid = true
                 logWarning(
                   s"Failed to extract state value watermark from condition $exprToCollectFrom " +

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

@@ -103,7 +103,7 @@ object TimeWindow {
    */
   private def getIntervalInMicroSeconds(interval: String): Long = {
     val cal = CalendarInterval.fromCaseInsensitiveString(interval)
-    if (cal.months > 0) {
+    if (cal.months != 0) {
       throw new IllegalArgumentException(
         s"Intervals greater than a month is not supported ($interval).")
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/EventTimeWatermark.scala

@@ -29,8 +29,7 @@ object EventTimeWatermark {
 
   def getDelayMs(delay: CalendarInterval): Long = {
     // We define month as `31 days` to simplify calculation.
-    val millisPerMonth = TimeUnit.MICROSECONDS.toMillis(CalendarInterval.MICROS_PER_DAY) * 31
-    delay.milliseconds + delay.months * millisPerMonth
+    delay.getDuration(31, TimeUnit.MILLISECONDS)
   }
 }
 

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

@@ -731,7 +731,7 @@ class Dataset[T] private[sql](
             s"Unable to parse time delay '$delayThreshold'",
             cause = Some(e))
       }
-    require(parsedDelay.milliseconds >= 0 && parsedDelay.months >= 0,
+    require(!parsedDelay.isNegative(31),
       s"delay threshold ($delayThreshold) should not be negative.")
     EliminateEventTimeWatermark(
       EventTimeWatermark(UnresolvedAttribute(eventTime), parsedDelay, logicalPlan))

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/GroupStateImpl.scala

@@ -160,12 +160,12 @@ private[sql] class GroupStateImpl[S] private(
 
   private def parseDuration(duration: String): Long = {
     val cal = CalendarInterval.fromCaseInsensitiveString(duration)
-    if (cal.milliseconds < 0 || cal.months < 0) {
-      throw new IllegalArgumentException(s"Provided duration ($duration) is not positive")
+    val daysPerMonth = 31
+    if (cal.isNegative(daysPerMonth)) {
+      throw new IllegalArgumentException(s"Provided duration ($duration) is negative")
     }
 
-    val millisPerMonth = TimeUnit.MICROSECONDS.toMillis(CalendarInterval.MICROS_PER_DAY) * 31
-    cal.milliseconds + cal.months * millisPerMonth
+    cal.getDuration(daysPerMonth, TimeUnit.MILLISECONDS)
   }
 
   private def checkTimeoutTimestampAllowed(): Unit = {

sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/Triggers.scala

@@ -31,7 +31,7 @@ private object Triggers {
 
   def convert(interval: String): Long = {
     val cal = CalendarInterval.fromCaseInsensitiveString(interval)
-    if (cal.months > 0) {
+    if (cal.months != 0) {
       throw new IllegalArgumentException(s"Doesn't support month or year interval: $interval")
     }
     TimeUnit.MICROSECONDS.toMillis(cal.microseconds)