[KYUUBI #4639] Support to specify confOverlay when executing statement with RESTful API

### _Why are the changes needed?_

As title.

With this pr, customer can execute SCALA code with `confOverlay`.
```
kyuubi.operation.language=SCALA
```
execute PYTHON code with
```
kyuubi.operation.language=PYTHON
```
### _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

- [ ] [Run test](https://kyuubi.readthedocs.io/en/master/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #4639 from turboFei/init_scala.

Closes #4639

cdf828f9a [fwang12] add ut
f4f2bc883 [fwang12] doc
2fe8a1659 [fwang12] save
f840cb4d9 [fwang12] conf overlay
eb49537ea [fwang12] conf overlay

Authored-by: fwang12 <fwang12@ebay.com>
Signed-off-by: fwang12 <fwang12@ebay.com>

docs/client/rest/rest_api.md

@@ -110,11 +110,12 @@ Create an operation with EXECUTE_STATEMENT type
 
 #### Request Body
 
-| Name         | Description                                                    | Type    |
-|:-------------|:---------------------------------------------------------------|:--------|
-| statement    | The SQL statement that you execute                             | String  |
-| runAsync     | The flag indicates whether the query runs synchronously or not | Boolean |
-| queryTimeout | The interval of query time out                                 | Long    |
+| Name         | Description                                                    | Type           |
+|:-------------|:---------------------------------------------------------------|:---------------|
+| statement    | The SQL statement that you execute                             | String         |
+| runAsync     | The flag indicates whether the query runs synchronously or not | Boolean        |
+| queryTimeout | The interval of query time out                                 | Long           |
+| confOverlay  | The conf to overlay only for current operation                 | Map of key=val |
 
 #### Response Body
 

docs/deployment/migration-guide.md

@@ -20,6 +20,7 @@
 ## Upgrading from Kyuubi 1.7.0 to 1.7.1
 
 * Since Kyuubi 1.7.1, `protocolVersion` is removed from the request parameters of the REST API `Open(create) a session`. All removed or unknown parameters will be silently ignored and affects nothing.
+* Since Kyuubi 1.7.1, `confOverlay` is supported in the request parameters of the REST API `Create an operation with EXECUTE_STATEMENT type`.
 
 ## Upgrading from Kyuubi 1.6 to 1.7
 

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/api/v1/dto/StatementRequest.java

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.client.api.v1.dto;
 
+import java.util.Collections;
+import java.util.Map;
 import java.util.Objects;
 import org.apache.commons.lang3.builder.ReflectionToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
@@ -25,13 +27,20 @@ public class StatementRequest {
   private String statement;
   private boolean runAsync;
   private Long queryTimeout;
+  private Map<String, String> confOverlay;
 
   public StatementRequest() {}
 
   public StatementRequest(String statement, boolean runAsync, Long queryTimeout) {
+    this(statement, runAsync, queryTimeout, Collections.emptyMap());
+  }
+
+  public StatementRequest(
+      String statement, boolean runAsync, Long queryTimeout, Map<String, String> confOverlay) {
     this.statement = statement;
     this.runAsync = runAsync;
     this.queryTimeout = queryTimeout;
+    this.confOverlay = confOverlay;
   }
 
   public String getStatement() {
@@ -58,6 +67,17 @@ public class StatementRequest {
     this.queryTimeout = queryTimeout;
   }
 
+  public Map<String, String> getConfOverlay() {
+    if (confOverlay == null) {
+      return Collections.emptyMap();
+    }
+    return confOverlay;
+  }
+
+  public void setConfOverlay(Map<String, String> confOverlay) {
+    this.confOverlay = confOverlay;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;

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

@@ -191,7 +191,7 @@ private[v1] class SessionsResource extends ApiRequestContext with Logging {
       fe.be.executeStatement(
         sessionHandleStr,
         request.getStatement,
-        Map.empty,
+        request.getConfOverlay.asScala.toMap,
         request.isRunAsync,
         request.getQueryTimeout)
     } catch {

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

@@ -19,7 +19,7 @@ package org.apache.kyuubi.server.api.v1
 
 import java.nio.charset.StandardCharsets
 import java.util
-import java.util.Base64
+import java.util.{Base64, Collections}
 import javax.ws.rs.client.Entity
 import javax.ws.rs.core.{GenericType, MediaType, Response}
 
@@ -192,7 +192,7 @@ class SessionsResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
 
     val pathPrefix = s"api/v1/sessions/$sessionHandle"
 
-    val statementReq = new StatementRequest("show tables", true, 3000)
+    var statementReq = new StatementRequest("show tables", true, 3000)
     response = webTarget
       .path(s"$pathPrefix/operations/statement").request(MediaType.APPLICATION_JSON_TYPE)
       .post(Entity.entity(statementReq, MediaType.APPLICATION_JSON_TYPE))
@@ -200,6 +200,18 @@ class SessionsResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
     var operationHandle = response.readEntity(classOf[OperationHandle])
     assert(operationHandle !== null)
 
+    statementReq = new StatementRequest(
+      "spark.sql(\"show tables\")",
+      true,
+      3000,
+      Collections.singletonMap(KyuubiConf.OPERATION_LANGUAGE.key, "SCALA"))
+    response = webTarget
+      .path(s"$pathPrefix/operations/statement").request(MediaType.APPLICATION_JSON_TYPE)
+      .post(Entity.entity(statementReq, MediaType.APPLICATION_JSON_TYPE))
+    assert(200 == response.getStatus)
+    operationHandle = response.readEntity(classOf[OperationHandle])
+    assert(operationHandle !== null)
+
     response = webTarget.path(s"$pathPrefix/operations/typeInfo").request()
       .post(Entity.entity(null, MediaType.APPLICATION_JSON_TYPE))
     assert(200 == response.getStatus)