Set default daysPerMonth to 31

Author: MaxGekk

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

@@ -30,7 +30,7 @@ object EventTimeWatermark {
 
   def getDelayMs(delay: CalendarInterval): Long = {
     // We define month as `31 days` to simplify calculation.
-    IntervalUtils.getDuration(delay, 31, TimeUnit.MILLISECONDS)
+    IntervalUtils.getDuration(delay, TimeUnit.MILLISECONDS)
   }
 }
 

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

@@ -95,14 +95,14 @@ object IntervalUtils {
    * Gets interval duration
    *
    * @param cal - the interval to get duration
-   * @param daysPerMonth - the number of days per one month
    * @param targetUnit - time units of the result
+   * @param daysPerMonth - the number of days per one month
    * @return duration in the specified time units
    */
   def getDuration(
       cal: CalendarInterval,
-      daysPerMonth: Int,
-      targetUnit: TimeUnit): Long = {
+      targetUnit: TimeUnit,
+      daysPerMonth: Int = 31): Long = {
     val monthsDuration = Math.multiplyExact(daysPerMonth * DateTimeUtils.MICROS_PER_DAY, cal.months)
     val result = Math.addExact(cal.microseconds, monthsDuration)
     targetUnit.convert(result, TimeUnit.MICROSECONDS)
@@ -115,7 +115,7 @@ object IntervalUtils {
    * @param daysPerMonth - the number of days per one month
    * @return true if duration of the given interval is less than 0 otherwise false
    */
-  def isNegative(cal: CalendarInterval, daysPerMonth: Int): Boolean = {
-    getDuration(cal, daysPerMonth, TimeUnit.MICROSECONDS) < 0
+  def isNegative(cal: CalendarInterval, daysPerMonth: Int = 31): Boolean = {
+    getDuration(cal, TimeUnit.MICROSECONDS, daysPerMonth) < 0
   }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/IntervalUtilsSuite.scala

@@ -27,17 +27,17 @@ import org.apache.spark.unsafe.types.CalendarInterval
 
 class IntervalUtilsSuite extends SparkFunSuite with Matchers {
   test("interval duration") {
-    def duration(s: String, daysPerMonth: Int, unit: TimeUnit): Long = {
-      getDuration(CalendarInterval.fromString(s), daysPerMonth, unit)
+    def duration(s: String, unit: TimeUnit, daysPerMonth: Int): Long = {
+      getDuration(CalendarInterval.fromString(s), unit, daysPerMonth)
     }
 
-    assert(duration("0 seconds", 31, TimeUnit.MILLISECONDS) === 0)
-    assert(duration("1 month", 31, TimeUnit.DAYS) === 31)
-    assert(duration("1 microsecond", 30, TimeUnit.MICROSECONDS) === 1)
-    assert(duration("1 month -30 days", 31, TimeUnit.DAYS) === 1)
+    assert(duration("0 seconds", TimeUnit.MILLISECONDS, 31) === 0)
+    assert(duration("1 month", TimeUnit.DAYS, 31) === 31)
+    assert(duration("1 microsecond", TimeUnit.MICROSECONDS, 30) === 1)
+    assert(duration("1 month -30 days", TimeUnit.DAYS, 31) === 1)
 
     try {
-      duration(Integer.MAX_VALUE + " month", 31, TimeUnit.SECONDS)
+      duration(Integer.MAX_VALUE + " month", TimeUnit.SECONDS, 31)
       fail("Expected to throw an exception for the invalid input")
     } catch {
       case e: ArithmeticException =>

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

@@ -732,7 +732,7 @@ class Dataset[T] private[sql](
             s"Unable to parse time delay '$delayThreshold'",
             cause = Some(e))
       }
-    require(!IntervalUtils.isNegative(parsedDelay, 31),
+    require(!IntervalUtils.isNegative(parsedDelay),
       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

@@ -161,12 +161,11 @@ private[sql] class GroupStateImpl[S] private(
 
   private def parseDuration(duration: String): Long = {
     val cal = CalendarInterval.fromCaseInsensitiveString(duration)
-    val daysPerMonth = 31
-    if (IntervalUtils.isNegative(cal, daysPerMonth)) {
+    if (IntervalUtils.isNegative(cal)) {
       throw new IllegalArgumentException(s"Provided duration ($duration) is negative")
     }
 
-    IntervalUtils.getDuration(cal, daysPerMonth, TimeUnit.MILLISECONDS)
+    IntervalUtils.getDuration(cal, TimeUnit.MILLISECONDS)
   }
 
   private def checkTimeoutTimestampAllowed(): Unit = {