[KYUUBI apache#4635] Support flink time type in query operation

Followup apache#1704, support flink `time` type in query operation

- [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.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request

Closes apache#4635 from yuruguo/support-flink-time-type.

Closes apache#4635

9f9a3e7 [Ruguo Yu] [Kyuubi apache#1704] Support flink time type in query operation

Authored-by: Ruguo Yu <jiang7chengzitc@163.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/operation/ExecuteStatement.scala

@@ -17,7 +17,7 @@
 
 package org.apache.kyuubi.engine.flink.operation
 
-import java.time.LocalDate
+import java.time.{LocalDate, LocalTime}
 import java.util
 
 import scala.collection.JavaConverters._
@@ -190,6 +190,9 @@ class ExecuteStatement(
         case _: DateType =>
           val date = RowSetUtils.formatLocalDate(LocalDate.ofEpochDay(r.getInt(i)))
           row.setField(i, date)
+        case _: TimeType =>
+          val time = RowSetUtils.formatLocalTime(LocalTime.ofNanoOfDay(r.getLong(i) * 1000 * 1000))
+          row.setField(i, time)
         case t: TimestampType =>
           val ts = RowSetUtils
             .formatLocalDateTime(r.getTimestamp(i, t.getPrecision)

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

@@ -307,6 +307,7 @@ object RowSet {
     case _: MapType => TTypeId.MAP_TYPE
     case _: RowType => TTypeId.STRUCT_TYPE
     case _: BinaryType => TTypeId.BINARY_TYPE
+    case _: TimeType => TTypeId.STRING_TYPE
     case t @ (_: ZonedTimestampType | _: LocalZonedTimestampType | _: MultisetType |
         _: YearMonthIntervalType | _: DayTimeIntervalType) =>
       throw new IllegalArgumentException(

externals/kyuubi-flink-sql-engine/src/test/scala/org/apache/kyuubi/engine/flink/operation/FlinkOperationSuite.scala

@@ -755,6 +755,20 @@ class FlinkOperationSuite extends WithFlinkSQLEngine with HiveJDBCTestHelper {
     }
   }
 
+  test("execute statement - select time") {
+    withJdbcStatement() { statement =>
+      val resultSet =
+        statement.executeQuery(
+          "select time '00:00:03', time '00:00:05.123456789'")
+      val metaData = resultSet.getMetaData
+      assert(metaData.getColumnType(1) === java.sql.Types.VARCHAR)
+      assert(metaData.getColumnType(2) === java.sql.Types.VARCHAR)
+      assert(resultSet.next())
+      assert(resultSet.getString(1) == "00:00:03")
+      assert(resultSet.getString(2) == "00:00:05.123")
+    }
+  }
+
   test("execute statement - select array") {
     withJdbcStatement() { statement =>
       val resultSet = statement.executeQuery("select array ['v1', 'v2', 'v3']")

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

@@ -18,7 +18,7 @@
 package org.apache.kyuubi.util
 
 import java.nio.ByteBuffer
-import java.time.{Instant, LocalDate, LocalDateTime, ZoneId}
+import java.time.{Instant, LocalDate, LocalDateTime, LocalTime, ZoneId}
 import java.time.chrono.IsoChronology
 import java.time.format.DateTimeFormatterBuilder
 import java.time.temporal.ChronoField
@@ -41,6 +41,12 @@ private[kyuubi] object RowSetUtils {
 
   private lazy val legacyDateFormatter = FastDateFormat.getInstance("yyyy-MM-dd", Locale.US)
 
+  private lazy val timeFormatter = createDateTimeFormatterBuilder()
+    .appendPattern("HH:mm:ss")
+    .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+    .toFormatter(Locale.US)
+    .withChronology(IsoChronology.INSTANCE)
+
   private lazy val timestampFormatter = createDateTimeFormatterBuilder()
     .appendPattern("yyyy-MM-dd HH:mm:ss")
     .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
@@ -59,6 +65,10 @@ private[kyuubi] object RowSetUtils {
     dateFormatter.format(ld)
   }
 
+  def formatLocalTime(lt: LocalTime): String = {
+    timeFormatter.format(lt)
+  }
+
   def formatLocalDateTime(ldt: LocalDateTime): String = {
     timestampFormatter.format(ldt)
   }