[SPARK-29983][SQL] Add an independent config for an optional INTERVAL clause

sql/catalyst/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBase.g4

@@ -49,6 +49,11 @@ grammar SqlBase;
    * When true, the behavior of keywords follows ANSI SQL standard.
    */
   public boolean SQL_standard_keyword_behavior = false;
+
+  /**
+   * When true, an INTERVAL keyword can be optional for interval values in SQL statements.
+   */
+  public boolean optional_interval_prefix = false;
 }
 
 singleStatement
@@ -788,7 +793,7 @@ booleanValue
 
 interval
     : INTERVAL (errorCapturingMultiUnitsInterval | errorCapturingUnitToUnitInterval)?
-    | {SQL_standard_keyword_behavior}? (errorCapturingMultiUnitsInterval | errorCapturingUnitToUnitInterval)
+    | {optional_interval_prefix}? (errorCapturingMultiUnitsInterval | errorCapturingUnitToUnitInterval)
     ;
 
 errorCapturingMultiUnitsInterval

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

@@ -96,11 +96,21 @@ abstract class AbstractSqlParser(conf: SQLConf) extends ParserInterface with Log
       case Dialect.SPARK => conf.dialectSparkAnsiEnabled
     }
 
+    // PostgreSQL cannot make INTERVAL keywords optional.
+    // In Spark dialect with setting `spark.sql.dialect.spark.ansi.enabled=true` and
+    // `spark.sql.parser.optionalIntervalPrefix=true`, the parser can omit
+    // INTERVAL keywords in SQL statements.
+    val optionalIntervalPrefix = conf.dialect match {
+      case Dialect.POSTGRESQL => false
+      case Dialect.SPARK => conf.dialectSparkAnsiEnabled && conf.optionalIntervalPrefix
+    }
+
     val lexer = new SqlBaseLexer(new UpperCaseCharStream(CharStreams.fromString(command)))
     lexer.removeErrorListeners()
     lexer.addErrorListener(ParseErrorListener)
     lexer.legacy_setops_precedence_enbled = conf.setOpsPrecedenceEnforced
     lexer.SQL_standard_keyword_behavior = SQLStandardKeywordBehavior
+    lexer.optional_interval_prefix = optionalIntervalPrefix
 
     val tokenStream = new CommonTokenStream(lexer)
     val parser = new SqlBaseParser(tokenStream)
@@ -109,6 +119,7 @@ abstract class AbstractSqlParser(conf: SQLConf) extends ParserInterface with Log
     parser.addErrorListener(ParseErrorListener)
     parser.legacy_setops_precedence_enbled = conf.setOpsPrecedenceEnforced
     parser.SQL_standard_keyword_behavior = SQLStandardKeywordBehavior
+    parser.optional_interval_prefix = optionalIntervalPrefix
 
     try {
       try {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1787,6 +1787,13 @@ object SQLConf {
     val SQL_STANDARD, ISO_8601, MULTI_UNITS = Value
   }
 
+  val OPTIONAL_INTERVAL_PREFIX =
+    buildConf("spark.sql.parser.optionalIntervalPrefix")
+      .doc("When true, an INTERVAL keyword can be optional for interval values in SQL statements," +
+        " e.g., 'SELECT 1 day'.")
+      .booleanConf
+      .createWithDefault(false)
+
   val INTERVAL_STYLE = buildConf("spark.sql.intervalOutputStyle")
     .doc("When converting interval values to strings (i.e. for display), this config decides the" +
       " interval string format. The value SQL_STANDARD will produce output matching SQL standard" +
@@ -2513,6 +2520,8 @@ class SQLConf extends Serializable with Logging {
   def storeAssignmentPolicy: StoreAssignmentPolicy.Value =
     StoreAssignmentPolicy.withName(getConf(STORE_ASSIGNMENT_POLICY))
 
+  def optionalIntervalPrefix: Boolean = getConf(OPTIONAL_INTERVAL_PREFIX)
+
   def intervalOutputStyle: IntervalStyle.Value = IntervalStyle.withName(getConf(INTERVAL_STYLE))
 
   def dialect: Dialect.Value = Dialect.withName(getConf(DIALECT))

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/ExpressionParserSuite.scala

@@ -615,8 +615,10 @@ class ExpressionParserSuite extends AnalysisTest {
       ).foreach { case (sign, expectedLiteral) =>
         assertEqual(s"${sign}interval $intervalValue", expectedLiteral)
 
-        // SPARK-23264 Support interval values without INTERVAL clauses if ANSI SQL enabled
-        withSQLConf(SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "true") {
+        // Checks if we can make INTERVAL optional
+        withSQLConf(
+            SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "true",
+            SQLConf.OPTIONAL_INTERVAL_PREFIX.key -> "true") {
           assertEqual(intervalValue, expected)
         }
       }
@@ -703,17 +705,23 @@ class ExpressionParserSuite extends AnalysisTest {
 
   test("SPARK-23264 Interval Compatibility tests") {
     def checkIntervals(intervalValue: String, expected: Literal): Unit = {
-      withSQLConf(SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "true") {
+      withSQLConf(
+          SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "true",
+          SQLConf.OPTIONAL_INTERVAL_PREFIX.key -> "true") {
         assertEqual(intervalValue, expected)
       }
 
       // Compatibility tests: If ANSI SQL disabled, `intervalValue` should be parsed as an alias
-      withSQLConf(SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "false") {
-        val aliases = defaultParser.parseExpression(intervalValue).collect {
-          case a @ Alias(_: Literal, name)
-            if intervalUnits.exists { unit => name.startsWith(unit.toString) } => a
+      Seq("true", "false").foreach { optionalIntervalPrefix =>
+        withSQLConf(
+            SQLConf.DIALECT_SPARK_ANSI_ENABLED.key -> "false",
+            SQLConf.OPTIONAL_INTERVAL_PREFIX.key -> optionalIntervalPrefix) {
+          val aliases = defaultParser.parseExpression(intervalValue).collect {
+            case a @ Alias(_: Literal, name)
+              if intervalUnits.exists { unit => name.startsWith(unit.toString) } => a
+          }
+          assert(aliases.size === 1)
         }
-        assert(aliases.size === 1)
       }
     }
     val forms = Seq("", "s")

sql/core/src/test/resources/sql-tests/inputs/ansi/interval.sql

@@ -1,17 +1,6 @@
+--SET spark.sql.parser.optionalIntervalPrefix=false
 --IMPORT interval.sql
 
--- the `interval` keyword can be omitted with ansi mode
-select 1 year 2 days;
-select '10-9' year to month;
-select '20 15:40:32.99899999' day to second;
-select 30 day day;
-select date'2012-01-01' - '2-2' year to month;
-select 1 month - 1 day;
-
--- malformed interval literal with ansi mode
-select 1 year to month;
-select '1' year to second;
-select 1 year '2-1' year to month;
-select (-30) day;
-select (a + 1) day;
-select 30 day day day;
+-- Cannot make INTERVAL keywords optional with the ANSI mode enabled and
+-- `spark.sql.parser.optionalIntervalPrefix=false`.
+--IMPORT ansi/optional-interval.sql

sql/core/src/test/resources/sql-tests/inputs/ansi/optional-interval.sql

@@ -0,0 +1,20 @@
+--SET spark.sql.parser.optionalIntervalPrefix=true
+--IMPORT interval.sql
+
+-- the `interval` keyword can be omitted with ansi mode enabled and
+-- `spark.sql.parser.optionalIntervalPrefix=true`.
+select 1 year 2 days;
+select '10-9' year to month;
+select '20 15:40:32.99899999' day to second;
+select 30 day day;
+select date'2012-01-01' - '2-2' year to month;
+select 1 month - 1 day;
+
+-- malformed interval literal with ansi mode enabled and
+-- `spark.sql.parser.optionalIntervalPrefix=true`.
+select 1 year to month;
+select '1' year to second;
+select 1 year '2-1' year to month;
+select (-30) day;
+select (a + 1) day;
+select 30 day day day;

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

@@ -1118,25 +1118,43 @@ struct<(INTERVAL '99 days 11 hours 22 minutes 33.123456 seconds' + INTERVAL '10
 -- !query 118
 select 1 year 2 days
 -- !query 118 schema
-struct<INTERVAL '1 years 2 days':interval>
+struct<>
 -- !query 118 output
-1 years 2 days
+org.apache.spark.sql.catalyst.parser.ParseException
+
+no viable alternative at input '1'(line 1, pos 7)
+
+== SQL ==
+select 1 year 2 days
+-------^^^
 
 
 -- !query 119
 select '10-9' year to month
 -- !query 119 schema
-struct<INTERVAL '10 years 9 months':interval>
+struct<>
 -- !query 119 output
-10 years 9 months
+org.apache.spark.sql.catalyst.parser.ParseException
+
+no viable alternative at input ''10-9''(line 1, pos 7)
+
+== SQL ==
+select '10-9' year to month
+-------^^^
 
 
 -- !query 120
 select '20 15:40:32.99899999' day to second
 -- !query 120 schema
-struct<INTERVAL '20 days 15 hours 40 minutes 32.998999 seconds':interval>
+struct<>
 -- !query 120 output
-20 days 15 hours 40 minutes 32.998999 seconds
+org.apache.spark.sql.catalyst.parser.ParseException
+
+no viable alternative at input ''20 15:40:32.99899999''(line 1, pos 7)
+
+== SQL ==
+select '20 15:40:32.99899999' day to second
+-------^^^
 
 
 -- !query 121
@@ -1146,27 +1164,39 @@ struct<>
 -- !query 121 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-no viable alternative at input 'day'(line 1, pos 14)
+no viable alternative at input '30'(line 1, pos 7)
 
 == SQL ==
 select 30 day day
---------------^^^
+-------^^^
 
 
 -- !query 122
 select date'2012-01-01' - '2-2' year to month
 -- !query 122 schema
-struct<CAST(CAST(DATE '2012-01-01' AS TIMESTAMP) - INTERVAL '2 years 2 months' AS DATE):date>
+struct<>
 -- !query 122 output
-2009-11-01
+org.apache.spark.sql.catalyst.parser.ParseException
+
+no viable alternative at input ''2-2''(line 1, pos 26)
+
+== SQL ==
+select date'2012-01-01' - '2-2' year to month
+--------------------------^^^
 
 
 -- !query 123
 select 1 month - 1 day
 -- !query 123 schema
-struct<INTERVAL '1 months -1 days':interval>
+struct<>
 -- !query 123 output
-1 months -1 days
+org.apache.spark.sql.catalyst.parser.ParseException
+
+no viable alternative at input '1'(line 1, pos 7)
+
+== SQL ==
+select 1 month - 1 day
+-------^^^
 
 
 -- !query 124
@@ -1176,7 +1206,7 @@ struct<>
 -- !query 124 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-The value of from-to unit must be a string(line 1, pos 7)
+no viable alternative at input '1'(line 1, pos 7)
 
 == SQL ==
 select 1 year to month
@@ -1190,7 +1220,7 @@ struct<>
 -- !query 125 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-Intervals FROM year TO second are not supported.(line 1, pos 7)
+no viable alternative at input ''1''(line 1, pos 7)
 
 == SQL ==
 select '1' year to second
@@ -1204,11 +1234,11 @@ struct<>
 -- !query 126 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-Can only have a single from-to unit in the interval literal syntax(line 1, pos 14)
+no viable alternative at input '1'(line 1, pos 7)
 
 == SQL ==
 select 1 year '2-1' year to month
---------------^^^
+-------^^^
 
 
 -- !query 127
@@ -1246,8 +1276,8 @@ struct<>
 -- !query 129 output
 org.apache.spark.sql.catalyst.parser.ParseException
 
-no viable alternative at input 'day'(line 1, pos 14)
+no viable alternative at input '30'(line 1, pos 7)
 
 == SQL ==
 select 30 day day day
---------------^^^
+-------^^^