[WIP][SPARK-35076][SQL] Parse interval literals as ANSI intervals

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

@@ -2474,6 +2474,7 @@ case class Flatten(child: Expression) extends UnaryExpression with NullIntoleran
     copy(child = newChild)
 }
 
+// scalastyle:off line.size.limit line.contains.tab
 @ExpressionDescription(
   usage = """
     _FUNC_(start, stop, step) - Generates an array of elements from start to stop (inclusive),
@@ -2502,12 +2503,17 @@ case class Flatten(child: Expression) extends UnaryExpression with NullIntoleran
        [1,2,3,4,5]
       > SELECT _FUNC_(5, 1);
        [5,4,3,2,1]
+      > SET spark.sql.legacy.interval.enabled=true;
+       spark.sql.legacy.interval.enabled	true
       > SELECT _FUNC_(to_date('2018-01-01'), to_date('2018-03-01'), interval 1 month);
        [2018-01-01,2018-02-01,2018-03-01]
+      > SET spark.sql.legacy.interval.enabled=false;
+       spark.sql.legacy.interval.enabled	false
   """,
   group = "array_funcs",
   since = "2.4.0"
 )
+// scalastyle:on line.size.limit line.contains.tab
 case class Sequence(
     start: Expression,
     stop: Expression,

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

@@ -2392,6 +2392,7 @@ object DatePart {
   }
 }
 
+// scalastyle:off line.size.limit line.contains.tab
 // scalastyle:off line.size.limit
 @ExpressionDescription(
   usage = "_FUNC_(field, source) - Extracts a part of the date/timestamp or interval source.",
@@ -2410,17 +2411,22 @@ object DatePart {
        224
       > SELECT _FUNC_('SECONDS', timestamp'2019-10-01 00:00:01.000001');
        1.000001
+      > SET spark.sql.legacy.interval.enabled=true;
+       spark.sql.legacy.interval.enabled	true
       > SELECT _FUNC_('days', interval 1 year 10 months 5 days);
        5
       > SELECT _FUNC_('seconds', interval 5 hours 30 seconds 1 milliseconds 1 microseconds);
        30.001001
+      > SET spark.sql.legacy.interval.enabled=false;
+       spark.sql.legacy.interval.enabled	false
   """,
   note = """
     The _FUNC_ function is equivalent to the SQL-standard function `EXTRACT(field FROM source)`
   """,
   group = "datetime_funcs",
   since = "3.0.0")
 // scalastyle:on line.size.limit
+// scalastyle:on line.size.limit line.contains.tab
 case class DatePart(field: Expression, source: Expression, child: Expression)
   extends RuntimeReplaceable {
 
@@ -2437,6 +2443,7 @@ case class DatePart(field: Expression, source: Expression, child: Expression)
     copy(child = newChild)
 }
 
+// scalastyle:off line.size.limit line.contains.tab
 // scalastyle:off line.size.limit
 @ExpressionDescription(
   usage = "_FUNC_(field FROM source) - Extracts a part of the date/timestamp or interval source.",
@@ -2475,17 +2482,22 @@ case class DatePart(field: Expression, source: Expression, child: Expression)
        224
       > SELECT _FUNC_(SECONDS FROM timestamp'2019-10-01 00:00:01.000001');
        1.000001
+      > SET spark.sql.legacy.interval.enabled=true;
+       spark.sql.legacy.interval.enabled	true
       > SELECT _FUNC_(days FROM interval 1 year 10 months 5 days);
        5
       > SELECT _FUNC_(seconds FROM interval 5 hours 30 seconds 1 milliseconds 1 microseconds);
        30.001001
+      > SET spark.sql.legacy.interval.enabled=false;
+       spark.sql.legacy.interval.enabled	false
   """,
   note = """
     The _FUNC_ function is equivalent to `date_part(field, source)`.
   """,
   group = "datetime_funcs",
   since = "3.0.0")
 // scalastyle:on line.size.limit
+// scalastyle:on line.size.limit line.contains.tab
 case class Extract(field: Expression, source: Expression, child: Expression)
   extends RuntimeReplaceable {
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.parser
 
 import java.util.Locale
+import java.util.concurrent.TimeUnit
 import javax.xml.bind.DatatypeConverter
 
 import scala.collection.JavaConverters._
@@ -2306,12 +2307,24 @@ class AstBuilder extends SqlBaseBaseVisitor[AnyRef] with SQLConfHelper with Logg
   }
 
   /**
-   * Create a [[CalendarInterval]] literal expression. Two syntaxes are supported:
+   * Create a [[CalendarInterval]] or ANSI interval literal expression.
+   * Two syntaxes are supported:
    * - multiple unit value pairs, for instance: interval 2 months 2 days.
    * - from-to unit, for instance: interval '1-2' year to month.
    */
   override def visitInterval(ctx: IntervalContext): Literal = withOrigin(ctx) {
-    Literal(parseIntervalLiteral(ctx), CalendarIntervalType)
+    val parsedInterval = parseIntervalLiteral(ctx)
+    if (SQLConf.get.legacyIntervalEnabled) {
+      Literal(parsedInterval, CalendarIntervalType)
+    } else if (parsedInterval.months != 0) {
+      if (parsedInterval.days != 0 || parsedInterval.microseconds != 0) {
+        throw QueryParsingErrors.mixedIntervalError(ctx)
+      }
+      Literal(parsedInterval.months, YearMonthIntervalType)
+    } else {
+      val micros = IntervalUtils.getDuration(parsedInterval, TimeUnit.MICROSECONDS)
+      Literal(micros, DayTimeIntervalType)
+    }
   }
 
   /**

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryParsingErrors.scala

@@ -22,6 +22,7 @@ import org.antlr.v4.runtime.ParserRuleContext
 import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.parser.SqlBaseParser._
 import org.apache.spark.sql.catalyst.trees.Origin
+import org.apache.spark.sql.internal.SQLConf
 
 /**
  * Object for grouping all error messages of the query parsing.
@@ -367,4 +368,10 @@ object QueryParsingErrors {
     new ParseException("LOCAL is supported only with file: scheme", ctx)
   }
 
+  def mixedIntervalError(ctx: IntervalContext): Throwable = {
+    new ParseException(
+      "Mixing of year-month and day-time fields is not allowed. " +
+      s"Set '${SQLConf.LEGACY_INTERVAL_ENABLED.key}' to true to enable the legacy interval type " +
+      "which supports mixed fields.", ctx)
+  }
 }

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

@@ -291,49 +291,47 @@ select timestamp '2019-01-01中文'
 -- !query
 select timestamp'2011-11-11 11:11:11' + interval '2' day
 -- !query schema
-struct<TIMESTAMP '2011-11-11 11:11:11' + INTERVAL '2 days':timestamp>
+struct<TIMESTAMP '2011-11-11 11:11:11' + 172800000000:timestamp>
 -- !query output
 2011-11-13 11:11:11
 
 
 -- !query
 select timestamp'2011-11-11 11:11:11' - interval '2' day
 -- !query schema
-struct<TIMESTAMP '2011-11-11 11:11:11' - INTERVAL '2 days':timestamp>
+struct<TIMESTAMP '2011-11-11 11:11:11' - 172800000000:timestamp>
 -- !query output
 2011-11-09 11:11:11
 
 
 -- !query
 select date'2011-11-11 11:11:11' + interval '2' second
 -- !query schema
-struct<>
+struct<DATE '2011-11-11' + 2000000:timestamp>
 -- !query output
-java.lang.IllegalArgumentException
-requirement failed: Cannot add hours, minutes or seconds, milliseconds, microseconds to a date
+2011-11-11 00:00:02
 
 
 -- !query
 select date'2011-11-11 11:11:11' - interval '2' second
 -- !query schema
-struct<>
+struct<DATE '2011-11-11' - 2000000:timestamp>
 -- !query output
-java.lang.IllegalArgumentException
-requirement failed: Cannot add hours, minutes or seconds, milliseconds, microseconds to a date
+2011-11-10 23:59:58
 
 
 -- !query
 select '2011-11-11' - interval '2' day
 -- !query schema
-struct<2011-11-11 - INTERVAL '2 days':string>
+struct<2011-11-11 - 172800000000:string>
 -- !query output
 2011-11-09 00:00:00
 
 
 -- !query
 select '2011-11-11 11:11:11' - interval '2' second
 -- !query schema
-struct<2011-11-11 11:11:11 - INTERVAL '2 seconds':string>
+struct<2011-11-11 11:11:11 - 2000000:string>
 -- !query output
 2011-11-11 11:11:09
 
@@ -353,7 +351,7 @@ select 1 - interval '2' second
 struct<>
 -- !query output
 org.apache.spark.sql.AnalysisException
-cannot resolve '1 + (- INTERVAL '2 seconds')' due to data type mismatch: argument 1 requires timestamp type, however, '1' is of int type.; line 1 pos 7
+cannot resolve '1 + (- 2000000)' due to data type mismatch: argument 1 requires timestamp type, however, '1' is of int type.; line 1 pos 7
 
 
 -- !query