[SPARK-35017][SQL] Transfer ANSI intervals via Hive Thrift server

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/ColumnValue.java

@@ -168,10 +168,6 @@ public static TColumnValue toTColumnValue(TypeDescriptor typeDescriptor, Object
     case TIMESTAMP_TYPE:
       // SPARK-31859, SPARK-31861: converted to string already in SparkExecuteStatementOperation
       return stringValue((String)value);
-    case INTERVAL_YEAR_MONTH_TYPE:
-      return stringValue((HiveIntervalYearMonth) value);
-    case INTERVAL_DAY_TIME_TYPE:
-      return stringValue((HiveIntervalDayTime) value);
     case DECIMAL_TYPE:
       String plainStr = value == null ? null : ((BigDecimal)value).toPlainString();
       return stringValue(plainStr);
@@ -183,6 +179,8 @@ public static TColumnValue toTColumnValue(TypeDescriptor typeDescriptor, Object
     case STRUCT_TYPE:
     case UNION_TYPE:
     case USER_DEFINED_TYPE:
+    case INTERVAL_YEAR_MONTH_TYPE:
+    case INTERVAL_DAY_TIME_TYPE:
       return stringValue((String)value);
     case NULL_TYPE:
       return stringValue((String)value);

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala

@@ -120,7 +120,8 @@ private[hive] class SparkExecuteStatementOperation(
           (from.getAs[CalendarInterval](ordinal), CalendarIntervalType),
           false,
           timeFormatters)
-      case _: ArrayType | _: StructType | _: MapType | _: UserDefinedType[_] =>
+      case _: ArrayType | _: StructType | _: MapType | _: UserDefinedType[_] |
+          YearMonthIntervalType | DayTimeIntervalType =>
         to += toHiveString((from.get(ordinal), dataTypes(ordinal)), false, timeFormatters)
     }
   }
@@ -377,6 +378,8 @@ object SparkExecuteStatementOperation {
       val attrTypeString = field.dataType match {
         case NullType => "void"
         case CalendarIntervalType => StringType.catalogString
+        case YearMonthIntervalType => "interval_year_month"
+        case DayTimeIntervalType => "interval_day_time"
         case other => other.catalogString
       }
       new FieldSchema(field.name, attrTypeString, field.getComment.getOrElse(""))

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkGetTypeInfoOperation.scala

@@ -99,6 +99,7 @@ private[hive] object SparkGetTypeInfoUtil {
       TINYINT_TYPE, SMALLINT_TYPE, INT_TYPE, BIGINT_TYPE,
       FLOAT_TYPE, DOUBLE_TYPE, DECIMAL_TYPE,
       DATE_TYPE, TIMESTAMP_TYPE,
-      ARRAY_TYPE, MAP_TYPE, STRUCT_TYPE, CHAR_TYPE, VARCHAR_TYPE)
+      ARRAY_TYPE, MAP_TYPE, STRUCT_TYPE, CHAR_TYPE, VARCHAR_TYPE,
+      INTERVAL_YEAR_MONTH_TYPE, INTERVAL_DAY_TIME_TYPE)
   }
 }

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/SparkThriftServerProtocolVersionsSuite.scala

@@ -20,6 +20,7 @@ package org.apache.spark.sql.hive.thriftserver
 import java.sql.{Date, Timestamp}
 import java.util.{List => JList, Properties}
 
+import org.apache.hadoop.hive.common.`type`.HiveIntervalDayTime
 import org.apache.hive.jdbc.{HiveConnection, HiveQueryResultSet}
 import org.apache.hive.service.auth.PlainSaslHelper
 import org.apache.hive.service.cli.GetInfoType
@@ -458,5 +459,17 @@ class SparkThriftServerProtocolVersionsSuite extends HiveThriftServer2TestBase {
         }
       }
     }
+
+    test(s"SPARK-35017: $version get day-time interval type") {
+      testExecuteStatementWithProtocolVersion(
+        version, "SELECT date'2021-01-01' - date'2020-12-31' AS dt") { rs =>
+        assert(rs.next())
+        assert(rs.getObject(1) === new HiveIntervalDayTime(1, 0, 0, 0, 0))
+        val metaData = rs.getMetaData
+        assert(metaData.getColumnName(1) === "dt")
+        assert(metaData.getColumnTypeName(1) === "interval_day_time")
+        assert(metaData.getColumnType(1) === java.sql.Types.OTHER)
+      }
+    }
   }
 }