[SPARK-35085][SQL] Get columns operation should handle ANSI interval column properly

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

@@ -131,7 +131,7 @@ private[hive] class SparkGetColumnsOperation(
    */
   private def getColumnSize(typ: DataType): Option[Int] = typ match {
     case dt @ (BooleanType | _: NumericType | DateType | TimestampType |
-               CalendarIntervalType | NullType) =>
+               CalendarIntervalType | NullType | YearMonthIntervalType | DayTimeIntervalType) =>
       Some(dt.defaultSize)
     case CharType(n) => Some(n)
     case StructType(fields) =>
@@ -171,8 +171,8 @@ private[hive] class SparkGetColumnsOperation(
     case BooleanType => java.sql.Types.BOOLEAN
     case ByteType => java.sql.Types.TINYINT
     case ShortType => java.sql.Types.SMALLINT
-    case IntegerType => java.sql.Types.INTEGER
-    case LongType => java.sql.Types.BIGINT
+    case IntegerType | YearMonthIntervalType => java.sql.Types.INTEGER
+    case LongType | DayTimeIntervalType => java.sql.Types.BIGINT
     case FloatType => java.sql.Types.FLOAT
     case DoubleType => java.sql.Types.DOUBLE
     case _: DecimalType => java.sql.Types.DECIMAL

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

@@ -377,6 +377,60 @@ class SparkMetadataOperationSuite extends HiveThriftServer2TestBase {
     }
   }
 
+  test("SPARK-35085: Get columns operation should handle ANSI interval column properly") {
+    val viewName1 = "view_interval1"
+    val yearMonthDDL =
+      s"CREATE GLOBAL TEMP VIEW $viewName1 as select interval '1-1' year to month as i"
+
+    withJdbcStatement(viewName1) { statement =>
+      statement.execute(yearMonthDDL)
+      val data = statement.getConnection.getMetaData
+      val rowSet = data.getColumns("", "global_temp", viewName1, null)
+      while (rowSet.next()) {
+        assert(rowSet.getString("TABLE_CAT") === null)
+        assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
+        assert(rowSet.getString("TABLE_NAME") === viewName1)
+        assert(rowSet.getString("COLUMN_NAME") === "i")
+        assert(rowSet.getInt("DATA_TYPE") === java.sql.Types.INTEGER)
+        assert(rowSet.getString("TYPE_NAME").equalsIgnoreCase(YearMonthIntervalType.sql))
+        assert(rowSet.getInt("COLUMN_SIZE") === YearMonthIntervalType.defaultSize)
+        assert(rowSet.getInt("DECIMAL_DIGITS") === 0)
+        assert(rowSet.getInt("NUM_PREC_RADIX") === 0)
+        assert(rowSet.getInt("NULLABLE") === 0)
+        assert(rowSet.getString("REMARKS") === "")
+        assert(rowSet.getInt("ORDINAL_POSITION") === 0)
+        assert(rowSet.getString("IS_NULLABLE") === "YES")
+        assert(rowSet.getString("IS_AUTO_INCREMENT") === "NO")
+      }
+    }
+
+    val viewName2 = "view_interval2"
+    val dayTimeDDL =
+      s"CREATE GLOBAL TEMP VIEW $viewName2 as select interval '1 2:3:4.001' day to second as i"
+
+    withJdbcStatement(viewName2) { statement =>
+      statement.execute(dayTimeDDL)
+      val data = statement.getConnection.getMetaData
+      val rowSet = data.getColumns("", "global_temp", viewName2, null)
+      while (rowSet.next()) {
+        assert(rowSet.getString("TABLE_CAT") === null)
+        assert(rowSet.getString("TABLE_SCHEM") === "global_temp")
+        assert(rowSet.getString("TABLE_NAME") === viewName2)
+        assert(rowSet.getString("COLUMN_NAME") === "i")
+        assert(rowSet.getInt("DATA_TYPE") === java.sql.Types.BIGINT)
+        assert(rowSet.getString("TYPE_NAME").equalsIgnoreCase(DayTimeIntervalType.sql))
+        assert(rowSet.getInt("COLUMN_SIZE") === DayTimeIntervalType.defaultSize)
+        assert(rowSet.getInt("DECIMAL_DIGITS") === 0)
+        assert(rowSet.getInt("NUM_PREC_RADIX") === 0)
+        assert(rowSet.getInt("NULLABLE") === 0)
+        assert(rowSet.getString("REMARKS") === "")
+        assert(rowSet.getInt("ORDINAL_POSITION") === 0)
+        assert(rowSet.getString("IS_NULLABLE") === "YES")
+        assert(rowSet.getString("IS_AUTO_INCREMENT") === "NO")
+      }
+    }
+  }
+
   test("handling null in view for get columns operations") {
     val viewName = "view_null"
     val ddl = s"CREATE GLOBAL TEMP VIEW $viewName as select null as n"