[KYUUBI apache#4372] Support to return null value for OperationsResource rowset

### _Why are the changes needed?_

With restful api, for query `select cast(null as int) c1`

#### Before
the result is `0`

#### After
the result is `null`.
### _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.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request

Closes apache#4372 from turboFei/null_value.

Closes apache#4372

b15f1eb [fwang12] nit
5f16855 [fwang12] fix
45c60dd [fwang12] check is set

Authored-by: fwang12 <[email protected]>
Signed-off-by: fwang12 <[email protected]>

Author: turboFei

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/OperationsResource.scala

@@ -182,19 +182,47 @@ private[v1] class OperationsResource extends ApiRequestContext with Logging {
             i.getSetField.name(),
             i.getSetField match {
               case TColumnValue._Fields.STRING_VAL =>
-                i.getStringVal.getFieldValue(TStringValue._Fields.VALUE)
+                if (i.getStringVal.isSetValue) {
+                  i.getStringVal.getFieldValue(TStringValue._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.BOOL_VAL =>
-                i.getBoolVal.getFieldValue(TBoolValue._Fields.VALUE)
+                if (i.getBoolVal.isSetValue) {
+                  i.getBoolVal.getFieldValue(TBoolValue._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.BYTE_VAL =>
-                i.getByteVal.getFieldValue(TByteValue._Fields.VALUE)
+                if (i.getByteVal.isSetValue) {
+                  i.getByteVal.getFieldValue(TByteValue._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.DOUBLE_VAL =>
-                i.getDoubleVal.getFieldValue(TDoubleValue._Fields.VALUE)
+                if (i.getDoubleVal.isSetValue) {
+                  i.getDoubleVal.getFieldValue(TDoubleValue._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.I16_VAL =>
-                i.getI16Val.getFieldValue(TI16Value._Fields.VALUE)
+                if (i.getI16Val.isSetValue) {
+                  i.getI16Val.getFieldValue(TI16Value._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.I32_VAL =>
-                i.getI32Val.getFieldValue(TI32Value._Fields.VALUE)
+                if (i.getI32Val.isSetValue) {
+                  i.getI32Val.getFieldValue(TI32Value._Fields.VALUE)
+                } else {
+                  null
+                }
               case TColumnValue._Fields.I64_VAL =>
-                i.getI64Val.getFieldValue(TI64Value._Fields.VALUE)
+                if (i.getI64Val.isSetValue) {
+                  i.getI64Val.getFieldValue(TI64Value._Fields.VALUE)
+                } else {
+                  null
+                }
             })
         }).asJava)
       })

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/OperationsResourceSuite.scala

@@ -126,6 +126,30 @@ class OperationsResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper
     assert(logRowSet.getRowCount == 1)
   }
 
+  test("test get result row set with null value") {
+    val opHandleStr = getOpHandleStr(
+      s"""
+         |select
+         |cast(null as string) as c1,
+         |cast(null as boolean) as c2,
+         |cast(null as byte) as c3,
+         |cast(null as double) as c4,
+         |cast(null as short) as c5,
+         |cast(null as int) as c6,
+         |cast(null as bigint) as c7
+         |""".stripMargin)
+    checkOpState(opHandleStr, FINISHED)
+    val response = webTarget.path(
+      s"api/v1/operations/$opHandleStr/rowset")
+      .queryParam("maxrows", "2")
+      .queryParam("fetchorientation", "FETCH_NEXT")
+      .request(MediaType.APPLICATION_JSON).get()
+    assert(200 == response.getStatus)
+    val logRowSet = response.readEntity(classOf[ResultRowSet])
+    assert(logRowSet.getRows.asScala.head.getFields.asScala.forall(_.getValue == null))
+    assert(logRowSet.getRowCount == 1)
+  }
+
   def getOpHandleStr(statement: String = "show tables"): String = {
     val sessionHandle = fe.be.openSession(
       HIVE_CLI_SERVICE_PROTOCOL_V2,