[KYUUBI #1710] Support to specify OPERATION_LANGUAGE with TExecuteStatementReq confOverlay

<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/apache/incubator-kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->
Now kyuubi supports SQL and SCALA language.

Now the way to switch OPERATION_LANAUAGE:

```
// for SCALA
spark.sql("set kyuubi.operation.language=SCALA")

// for SQL
set kyuubi.operation.language=SCALA
```

It is more user friendly that user can specify the OPERATION_LANGUAGE for each ExecuteStatement.

### _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/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #1710 from turboFei/conf_overlay.

Closes #1710

385264d [Fei Wang] fix ut
58fbdbd [Fei Wang] fix npe
2718fb4 [Fei Wang] Support to get OPERATION_LANGUAGE from execute statement confOverlay

Authored-by: Fei Wang <fwang12@ebay.com>
Signed-off-by: Cheng Pan <chengpan@apache.org>

Author: turboFei

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

@@ -27,6 +27,7 @@ class FlinkSQLOperationManager extends OperationManager("FlinkSQLOperationManage
   override def newExecuteStatementOperation(
       session: Session,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): Operation = {
     val op = new ExecuteStatement(session, statement, runAsync, queryTimeout)

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkSQLOperationManager.scala

@@ -48,10 +48,12 @@ class SparkSQLOperationManager private (name: String) extends OperationManager(n
   override def newExecuteStatementOperation(
       session: Session,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): Operation = {
     val spark = session.asInstanceOf[SparkSessionImpl].spark
-    val lang = spark.conf.get(OPERATION_LANGUAGE.key, operationLanguageDefault)
+    val lang = confOverlay.get(OPERATION_LANGUAGE.key)
+      .getOrElse(spark.conf.get(OPERATION_LANGUAGE.key, operationLanguageDefault))
     val operation =
       OperationLanguages.withName(lang.toUpperCase(Locale.ROOT)) match {
         case OperationLanguages.SQL =>

kyuubi-common/src/main/scala/org/apache/kyuubi/operation/OperationManager.scala

@@ -46,6 +46,7 @@ abstract class OperationManager(name: String) extends AbstractService(name) {
   def newExecuteStatementOperation(
       session: Session,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): Operation
   def newGetTypeInfoOperation(session: Session): Operation

kyuubi-common/src/main/scala/org/apache/kyuubi/service/AbstractBackendService.scala

@@ -56,9 +56,14 @@ abstract class AbstractBackendService(name: String)
   override def executeStatement(
       sessionHandle: SessionHandle,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): OperationHandle = {
-    sessionManager.getSession(sessionHandle).executeStatement(statement, runAsync, queryTimeout)
+    sessionManager.getSession(sessionHandle).executeStatement(
+      statement,
+      confOverlay,
+      runAsync,
+      queryTimeout)
   }
 
   override def getTypeInfo(sessionHandle: SessionHandle): OperationHandle = {

kyuubi-common/src/main/scala/org/apache/kyuubi/service/BackendService.scala

@@ -48,6 +48,7 @@ trait BackendService {
   def executeStatement(
       sessionHandle: SessionHandle,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): OperationHandle
 

kyuubi-common/src/main/scala/org/apache/kyuubi/service/ThriftBinaryFrontendService.scala

@@ -261,9 +261,14 @@ abstract class ThriftBinaryFrontendService(name: String)
       val sessionHandle = SessionHandle(req.getSessionHandle)
       val statement = req.getStatement
       val runAsync = req.isRunAsync
-      // val confOverlay = req.getConfOverlay
+      val confOverlay = Option(req.getConfOverlay).getOrElse(Map.empty.asJava)
       val queryTimeout = req.getQueryTimeout
-      val operationHandle = be.executeStatement(sessionHandle, statement, runAsync, queryTimeout)
+      val operationHandle = be.executeStatement(
+        sessionHandle,
+        statement,
+        confOverlay.asScala.toMap,
+        runAsync,
+        queryTimeout)
       resp.setOperationHandle(operationHandle.toTOperationHandle)
       resp.setStatus(OK_STATUS)
     } catch {

kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala

@@ -115,10 +115,11 @@ abstract class AbstractSession(
 
   override def executeStatement(
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): OperationHandle = withAcquireRelease() {
     val operation = sessionManager.operationManager
-      .newExecuteStatementOperation(this, statement, runAsync, queryTimeout)
+      .newExecuteStatementOperation(this, statement, confOverlay, runAsync, queryTimeout)
     runOperation(operation)
   }
 

kyuubi-common/src/main/scala/org/apache/kyuubi/session/Session.scala

@@ -47,7 +47,11 @@ trait Session {
 
   def getInfo(infoType: TGetInfoType): TGetInfoValue
 
-  def executeStatement(statement: String, runAsync: Boolean, queryTimeout: Long): OperationHandle
+  def executeStatement(
+      statement: String,
+      confOverlay: Map[String, String],
+      runAsync: Boolean,
+      queryTimeout: Long): OperationHandle
 
   def getTableTypes: OperationHandle
   def getTypeInfo: OperationHandle

kyuubi-common/src/test/scala/org/apache/kyuubi/operation/NoopOperationManager.scala

@@ -31,6 +31,7 @@ class NoopOperationManager extends OperationManager("noop") {
   override def newExecuteStatementOperation(
       session: Session,
       statement: String,
+      confOverlay: Map[String, String],
       runAsync: Boolean,
       queryTimeout: Long): Operation = {
     val operation =

kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala

@@ -93,11 +93,13 @@ class KyuubiSyncThriftClient private (protocol: TProtocol)
 
   def executeStatement(
       statement: String,
+      confOverlay: Map[String, String],
       shouldRunAsync: Boolean,
       queryTimeout: Long): TOperationHandle = {
     val req = new TExecuteStatementReq()
     req.setSessionHandle(_remoteSessionHandle)
     req.setStatement(statement)
+    req.setConfOverlay(confOverlay.asJava)
     req.setRunAsync(shouldRunAsync)
     req.setQueryTimeout(queryTimeout)
     val resp = withLockAcquired(ExecuteStatement(req))