[KYUUBI #2207] Support newly added spark data types: DayTimeIntervalType/YearMonthIntervalType

### _Why are the changes needed?_
#2207
Support newly added datatypes
spark3.2 : DayTimeIntervalType/YearMonthIntervalType

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2212 from Nick-0723/2207.

Closes #2207

c08ee99 [Nick Song] revert DayTimeIntervalType's part
abc56be [Nick Song] support newly data types DayTimeIntervalType/YearMonthIntervalType/TimestampNTZType

Authored-by: Nick Song <chun2184@163.com>
Signed-off-by: Kent Yao <yao@apache.org>

Author: Nick-0723

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/IntervalQualifier.scala

@@ -20,7 +20,7 @@ package org.apache.kyuubi.engine.spark.schema
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
 import java.sql.Timestamp
-import java.time.{Duration, Instant, LocalDate, ZoneId}
+import java.time._
 import java.util.Date
 
 import scala.collection.JavaConverters._
@@ -29,7 +29,6 @@ import org.apache.hive.service.rpc.thrift._
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.types._
 
-import org.apache.kyuubi.engine.spark.schema.IntervalQualifier.{DAY, HOUR, MINUTE, SECOND}
 import org.apache.kyuubi.util.RowSetUtils._
 
 object RowSet {
@@ -255,16 +254,9 @@ object RowSet {
         // Only match string in nested type values
         "\"" + s + "\""
 
-      case (d: Duration, dt) =>
-        if (dt.simpleString == DAY.toString()) {
-          IntervalQualifier.toDayTimeIntervalString(d, DAY)
-        } else if (dt.simpleString == HOUR.toString()) {
-          IntervalQualifier.toDayTimeIntervalString(d, HOUR)
-        } else if (dt.simpleString == MINUTE.toString()) {
-          IntervalQualifier.toDayTimeIntervalString(d, MINUTE)
-        } else {
-          IntervalQualifier.toDayTimeIntervalString(d, SECOND)
-        }
+      case (d: Duration, _) => toDayTimeIntervalString(d)
+
+      case (p: Period, _) => toYearMonthIntervalString(p)
 
       case (seq: scala.collection.Seq[_], ArrayType(typ, _)) =>
         seq.map(v => (v, typ)).map(e => toHiveString(e, timeZone)).mkString("[", ",", "]")

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala

@@ -41,8 +41,10 @@ object SchemaHelper {
     case TimestampType => TTypeId.TIMESTAMP_TYPE
     case BinaryType => TTypeId.BINARY_TYPE
     case CalendarIntervalType => TTypeId.STRING_TYPE
-    case dt if dt.simpleString.startsWith("interval day") =>
+    case dt if dt.getClass.getSimpleName.equals("DayTimeIntervalType") =>
       TTypeId.INTERVAL_DAY_TIME_TYPE
+    case ym if ym.getClass.getSimpleName.equals("YearMonthIntervalType") =>
+      TTypeId.INTERVAL_YEAR_MONTH_TYPE
     case _: ArrayType => TTypeId.ARRAY_TYPE
     case _: MapType => TTypeId.MAP_TYPE
     case _: StructType => TTypeId.STRUCT_TYPE

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/SchemaHelper.scala

@@ -19,19 +19,24 @@ package org.apache.kyuubi.util
 
 import java.nio.ByteBuffer
 import java.sql.Timestamp
-import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
+import java.time.{Duration, Instant, LocalDate, LocalDateTime, Period, ZoneId}
 import java.time.chrono.IsoChronology
 import java.time.format.DateTimeFormatter
 import java.time.format.DateTimeFormatterBuilder
 import java.time.temporal.ChronoField
 import java.util.{Date, Locale}
+import java.util.concurrent.TimeUnit
 
 import scala.language.implicitConversions
 
 import org.apache.commons.lang3.time.FastDateFormat
 
 private[kyuubi] object RowSetUtils {
 
+  final private val SECOND_PER_MINUTE: Long = 60L
+  final private val SECOND_PER_HOUR: Long = SECOND_PER_MINUTE * 60L
+  final private val SECOND_PER_DAY: Long = SECOND_PER_HOUR * 24L
+
   private lazy val dateFormatter = {
     createDateTimeFormatterBuilder().appendPattern("yyyy-MM-dd")
       .toFormatter(Locale.US)
@@ -79,4 +84,27 @@ private[kyuubi] object RowSetUtils {
   implicit def bitSetToBuffer(bitSet: java.util.BitSet): ByteBuffer = {
     ByteBuffer.wrap(bitSet.toByteArray)
   }
+
+  def toDayTimeIntervalString(d: Duration): String = {
+    var rest = d.getSeconds
+    var sign = ""
+    if (d.getSeconds < 0) {
+      sign = "-"
+      rest = -rest
+    }
+    val days = TimeUnit.SECONDS.toDays(rest)
+    rest %= SECOND_PER_DAY
+    val hours = TimeUnit.SECONDS.toHours(rest)
+    rest %= SECOND_PER_HOUR
+    val minutes = TimeUnit.SECONDS.toMinutes(rest)
+    val seconds = rest % SECOND_PER_MINUTE
+    f"$sign$days $hours%02d:$minutes%02d:$seconds%02d.${d.getNano}%09d"
+  }
+
+  def toYearMonthIntervalString(d: Period): String = {
+    val years = d.getYears
+    val months = d.getMonths
+    val sign = if (years < 0 || months < 0) "-" else ""
+    s"$sign${Math.abs(years)}-${Math.abs(months)}"
+  }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/util/RowSetUtils.scala

@@ -24,10 +24,12 @@ import scala.collection.JavaConverters._
 import org.apache.commons.lang3.StringUtils
 import org.apache.hive.service.rpc.thrift.{TExecuteStatementReq, TFetchResultsReq, TOpenSessionReq, TStatusCode}
 
-import org.apache.kyuubi.KYUUBI_VERSION
+import org.apache.kyuubi.{KYUUBI_VERSION, Utils}
 
 trait SparkQueryTests extends HiveJDBCTestHelper {
 
+  protected lazy val SPARK_ENGINE_MAJOR_MINOR_VERSION: (Int, Int) = sparkEngineMajorMinorVersion
+
   test("execute statement - select null") {
     withJdbcStatement() { statement =>
       val resultSet = statement.executeQuery("SELECT NULL AS col")
@@ -175,25 +177,70 @@ trait SparkQueryTests extends HiveJDBCTestHelper {
   test("execute statement - select daytime interval") {
     withJdbcStatement() { statement =>
       Map(
-        "-interval 2 day" -> "-2 00:00:00.000000000",
-        "-interval 200 day" -> "-200 00:00:00.000000000",
-        "interval 1 day 1 hour" -> "1 01:00:00.000000000",
-        "interval 1 day 1 hour -60 minutes" -> "1 00:00:00.000000000",
-        "interval 1 day 1 hour -60 minutes 30 seconds" -> "1 00:00:30.000000000",
+        "interval 1 day 1 hour -60 minutes 30 seconds" ->
+          Tuple2("1 00:00:30.000000000", "1 days 30 seconds"),
+        "interval 30 seconds 12345 milliseconds" ->
+          Tuple2("0 00:00:42.345000000", "42.345 seconds"),
+        "interval 1 hour 59 minutes 30 seconds 12345 milliseconds" ->
+          Tuple2("0 01:59:42.345000000", "1 hours 59 minutes 42.345 seconds"),
+        "-interval 2 day" -> Tuple2("-2 00:00:00.000000000", "-2 days"),
+        "interval 59 minutes 30 seconds 12345 milliseconds" ->
+          Tuple2("0 00:59:42.345000000", "59 minutes 42.345 seconds"),
+        "interval 25 hour" -> Tuple2("1 01:00:00.000000000", "25 hours"),
+        "interval 1 hour 62 minutes" -> Tuple2("0 02:02:00.000000000", "2 hours 2 minutes"),
         "interval 1 day 1 hour 59 minutes 30 seconds 12345 milliseconds" ->
-          "1 01:59:42.345000000").foreach { kv => // value -> result pair
+          Tuple2("1 01:59:42.345000000", "1 days 1 hours 59 minutes 42.345 seconds"),
+        "interval 1 day 1 hour -60 minutes" -> Tuple2("1 00:00:00.000000000", "1 days"),
+        "INTERVAL 30 SECONDS" -> Tuple2("0 00:00:30.000000000", "30 seconds"),
+        "interval -60 minutes 30 seconds" ->
+          Tuple2("-0 00:59:30.000000000", "-59 minutes -30 seconds"),
+        "-interval 200 day" -> Tuple2("-200 00:00:00.000000000", "-200 days"),
+        "interval 1 hour -60 minutes 30 seconds" -> Tuple2("0 00:00:30.000000000", "30 seconds"),
+        "interval 62 minutes" -> Tuple2("0 01:02:00.000000000", "1 hours 2 minutes"),
+        "interval 1 day 1 hour" -> Tuple2("1 01:00:00.000000000", "1 days 1 hours")).foreach {
+        kv => // value -> result pair
+          val resultSet = statement.executeQuery(s"SELECT ${kv._1} AS col")
+          assert(resultSet.next())
+          val result = resultSet.getString("col")
+          val metaData = resultSet.getMetaData
+          if (SPARK_ENGINE_MAJOR_MINOR_VERSION._1 == 3
+            && SPARK_ENGINE_MAJOR_MINOR_VERSION._2 < 2) {
+            // for spark 3.1 and backwards
+            assert(result === kv._2._2)
+            assert(metaData.getPrecision(1) === Int.MaxValue)
+            assert(resultSet.getMetaData.getColumnType(1) === java.sql.Types.VARCHAR)
+          } else {
+            assert(result === kv._2._1)
+            assert(metaData.getPrecision(1) === 29)
+            assert(resultSet.getMetaData.getColumnType(1) === java.sql.Types.OTHER)
+          }
+          assert(metaData.getScale(1) === 0)
+      }
+    }
+  }
+
+  test("execute statement - select year/month interval") {
+    withJdbcStatement() { statement =>
+      Map(
+        "INTERVAL 2022 YEAR" -> Tuple2("2022-0", "2022 years"),
+        "INTERVAL '2021-07' YEAR TO MONTH" -> Tuple2("2021-7", "2021 years 7 months"),
+        "INTERVAL 3 MONTH" -> Tuple2("0-3", "3 months"),
+        "INTERVAL 241 MONTH" -> Tuple2("20-1", "20 years 1 months"),
+        "INTERVAL -1 year -25 MONTH" -> Tuple2("-3-1", "-3 years -1 months"),
+        "INTERVAL 3 year -25 MONTH" -> Tuple2("0-11", "11 months")).foreach { kv =>
         val resultSet = statement.executeQuery(s"SELECT ${kv._1} AS col")
         assert(resultSet.next())
         val result = resultSet.getString("col")
         val metaData = resultSet.getMetaData
-        if (result.contains("days")) {
+        if (SPARK_ENGINE_MAJOR_MINOR_VERSION._1 == 3
+          && SPARK_ENGINE_MAJOR_MINOR_VERSION._2 < 2) {
           // for spark 3.1 and backwards
-          assert(result.split("days").head.trim === kv._2.split(" ").head)
+          assert(result === kv._2._2)
           assert(metaData.getPrecision(1) === Int.MaxValue)
           assert(resultSet.getMetaData.getColumnType(1) === java.sql.Types.VARCHAR)
         } else {
-          assert(result === kv._2)
-          assert(metaData.getPrecision(1) === 29)
+          assert(result === kv._2._1)
+          assert(metaData.getPrecision(1) === 11)
           assert(resultSet.getMetaData.getColumnType(1) === java.sql.Types.OTHER)
         }
         assert(metaData.getScale(1) === 0)
@@ -545,4 +592,15 @@ trait SparkQueryTests extends HiveJDBCTestHelper {
       assert(foundOperationLangItem)
     }
   }
+
+  def sparkEngineMajorMinorVersion: (Int, Int) = {
+    var sparkRuntimeVer = ""
+    withJdbcStatement() { stmt =>
+      val result = stmt.executeQuery("SELECT version()")
+      assert(result.next())
+      sparkRuntimeVer = result.getString(1)
+      assert(!result.next())
+    }
+    Utils.majorMinorVersion(sparkRuntimeVer)
+  }
 }