You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by ch...@apache.org on 2022/09/08 02:21:20 UTC
[incubator-kyuubi] 02/02: [KYUUBI #3122] GetInfo supports return server/engine info
This is an automated email from the ASF dual-hosted git repository.
chengpan pushed a commit to branch branch-1.6
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
commit d9f1d0b25fbb0d0c9fa7c06e004cb03bd67bf551
Author: Cheng Pan <ch...@apache.org>
AuthorDate: Thu Sep 8 02:14:38 2022 +0000
[KYUUBI #3122] GetInfo supports return server/engine info
### _Why are the changes needed?_
Workaround for #3032, close #3323
There are known some ODBC drivers e.g. [Databricks ODBC driver](https://www.databricks.com/spark/odbc-drivers-download) depending on `TGetInfoType.CLI_DBMS_VER` and `TGetInfoType.CLI_DBMS_NAME` to check server compatibilities and abilities.
This PR proposes to introduce a new configuration `kyuubi.server.info.provider=SERVER/ENGINE` to make GetInfo support return either server or engine information.
Since beeline will call GetInfo in the initialization phase, to make sure the beeline fast open experience, in async launch mode, when the engine is not ready, return server info regardless `kyuubi.server.info.provider`.
### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [x] Add screenshots for manual tests if appropriate
Testing w/ PowerBI
- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #3122 from pan3793/info.
Closes #3122
742bdbe3 [Cheng Pan] nit
bb85d2ba [Cheng Pan] style
fd75238c [Cheng Pan] fix
9ddb2afd [Cheng Pan] nit
fd8f7979 [Cheng Pan] fix ut
840205e5 [Cheng Pan] nit
f9996d53 [Cheng Pan] GetInfo supports returning engine info
Authored-by: Cheng Pan <ch...@apache.org>
Signed-off-by: Cheng Pan <ch...@apache.org>
---
docs/deployment/settings.md | 1 +
.../engine/flink/session/FlinkSessionImpl.scala | 13 ++-
.../engine/hive/session/HiveSessionImpl.scala | 17 ++-
.../engine/spark/session/SparkSessionImpl.scala | 18 +++-
.../spark/operation/SparkOperationSuite.scala | 13 +++
.../it/flink/operation/FlinkOperationSuite.scala | 24 +++++
.../KyuubiOperationHiveEnginePerUserSuite.scala | 24 +++++
.../org/apache/kyuubi/config/KyuubiConf.scala | 10 ++
.../apache/kyuubi/session/AbstractSession.scala | 10 +-
.../kyuubi/operation/SparkMetadataTests.scala | 27 ++---
.../kyuubi/service/TFrontendServiceSuite.scala | 9 +-
.../apache/kyuubi/session/KyuubiSessionImpl.scala | 13 +--
.../KyuubiOperationPerConnectionSuite.scala | 4 +-
.../operation/KyuubiOperationPerUserSuite.scala | 119 +++++++++++++--------
14 files changed, 220 insertions(+), 82 deletions(-)
diff --git a/docs/deployment/settings.md b/docs/deployment/settings.md
index 6513ed012..6b7811500 100644
--- a/docs/deployment/settings.md
+++ b/docs/deployment/settings.md
@@ -416,6 +416,7 @@ kyuubi.operation.status.polling.timeout|PT5S|Timeout(ms) for long polling asynch
Key | Default | Meaning | Type | Since
--- | --- | --- | --- | ---
+kyuubi.server.info.provider|SERVER|The server information provider name, some clients may rely on this information to check the server compatibilities and functionalities. <li>SERVER: Return Kyuubi server information.</li> <li>ENGINE: Return Kyuubi engine information.</li>|string|1.6.1
kyuubi.server.limit.connections.per.ipaddress|<undefined>|Maximum kyuubi server connections per ipaddress. Any user exceeding this limit will not be allowed to connect.|int|1.6.0
kyuubi.server.limit.connections.per.user|<undefined>|Maximum kyuubi server connections per user. Any user exceeding this limit will not be allowed to connect.|int|1.6.0
kyuubi.server.limit.connections.per.user.ipaddress|<undefined>|Maximum kyuubi server connections per user:ipaddress combination. Any user-ipaddress exceeding this limit will not be allowed to connect.|int|1.6.0
diff --git a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/session/FlinkSessionImpl.scala b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/session/FlinkSessionImpl.scala
index 99d223f20..7c6107ee3 100644
--- a/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/session/FlinkSessionImpl.scala
+++ b/externals/kyuubi-flink-sql-engine/src/main/scala/org/apache/kyuubi/engine/flink/session/FlinkSessionImpl.scala
@@ -19,11 +19,13 @@ package org.apache.kyuubi.engine.flink.session
import scala.util.control.NonFatal
+import org.apache.flink.runtime.util.EnvironmentInformation
import org.apache.flink.table.client.gateway.SqlExecutionException
import org.apache.flink.table.client.gateway.context.SessionContext
import org.apache.flink.table.client.gateway.local.LocalExecutor
-import org.apache.hive.service.rpc.thrift.TProtocolVersion
+import org.apache.hive.service.rpc.thrift.{TGetInfoType, TGetInfoValue, TProtocolVersion}
+import org.apache.kyuubi.KyuubiSQLException
import org.apache.kyuubi.engine.flink.FlinkEngineUtils
import org.apache.kyuubi.session.{AbstractSession, SessionManager}
@@ -66,4 +68,13 @@ class FlinkSessionImpl(
}
super.open()
}
+
+ override def getInfo(infoType: TGetInfoType): TGetInfoValue = withAcquireRelease() {
+ infoType match {
+ case TGetInfoType.CLI_SERVER_NAME | TGetInfoType.CLI_DBMS_NAME =>
+ TGetInfoValue.stringValue("Apache Flink")
+ case TGetInfoType.CLI_DBMS_VER => TGetInfoValue.stringValue(EnvironmentInformation.getVersion)
+ case _ => throw KyuubiSQLException(s"Unrecognized GetInfoType value: $infoType")
+ }
+ }
}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
index f59a69cba..3b85f94df 100644
--- a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
@@ -21,9 +21,11 @@ import java.util.HashMap
import scala.collection.JavaConverters._
+import org.apache.hive.common.util.HiveVersionInfo
import org.apache.hive.service.cli.session.HiveSession
-import org.apache.hive.service.rpc.thrift.TProtocolVersion
+import org.apache.hive.service.rpc.thrift.{TGetInfoType, TGetInfoValue, TProtocolVersion}
+import org.apache.kyuubi.KyuubiSQLException
import org.apache.kyuubi.engine.hive.events.HiveSessionEvent
import org.apache.kyuubi.events.EventBus
import org.apache.kyuubi.operation.{Operation, OperationHandle}
@@ -54,6 +56,19 @@ class HiveSessionImpl(
super.runOperation(operation)
}
+ override def getInfo(infoType: TGetInfoType): TGetInfoValue = withAcquireRelease() {
+ infoType match {
+ case TGetInfoType.CLI_SERVER_NAME => TGetInfoValue.stringValue("Hive")
+ case TGetInfoType.CLI_DBMS_NAME => TGetInfoValue.stringValue("Apache Hive")
+ case TGetInfoType.CLI_DBMS_VER => TGetInfoValue.stringValue(HiveVersionInfo.getVersion)
+ case TGetInfoType.CLI_ODBC_KEYWORDS => TGetInfoValue.stringValue("Unimplemented")
+ case TGetInfoType.CLI_MAX_COLUMN_NAME_LEN |
+ TGetInfoType.CLI_MAX_SCHEMA_NAME_LEN |
+ TGetInfoType.CLI_MAX_TABLE_NAME_LEN => TGetInfoValue.lenValue(128)
+ case _ => throw KyuubiSQLException(s"Unrecognized GetInfoType value: $infoType")
+ }
+ }
+
override def close(): Unit = {
sessionEvent.endTime = System.currentTimeMillis()
EventBus.post(sessionEvent)
diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
index 6abbf4266..eda7028cb 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSessionImpl.scala
@@ -17,9 +17,10 @@
package org.apache.kyuubi.engine.spark.session
-import org.apache.hive.service.rpc.thrift.TProtocolVersion
+import org.apache.hive.service.rpc.thrift.{TGetInfoType, TGetInfoValue, TProtocolVersion}
import org.apache.spark.sql.{AnalysisException, SparkSession}
+import org.apache.kyuubi.KyuubiSQLException
import org.apache.kyuubi.engine.spark.events.SessionEvent
import org.apache.kyuubi.engine.spark.operation.SparkSQLOperationManager
import org.apache.kyuubi.engine.spark.shim.SparkCatalogShim
@@ -75,11 +76,24 @@ class SparkSessionImpl(
super.runOperation(operation)
}
+ override def getInfo(infoType: TGetInfoType): TGetInfoValue = withAcquireRelease() {
+ infoType match {
+ case TGetInfoType.CLI_SERVER_NAME | TGetInfoType.CLI_DBMS_NAME =>
+ TGetInfoValue.stringValue("Spark SQL")
+ case TGetInfoType.CLI_DBMS_VER => TGetInfoValue.stringValue(org.apache.spark.SPARK_VERSION)
+ case TGetInfoType.CLI_ODBC_KEYWORDS => TGetInfoValue.stringValue("Unimplemented")
+ case TGetInfoType.CLI_MAX_COLUMN_NAME_LEN |
+ TGetInfoType.CLI_MAX_SCHEMA_NAME_LEN |
+ TGetInfoType.CLI_MAX_TABLE_NAME_LEN => TGetInfoValue.lenValue(128)
+ case _ => throw KyuubiSQLException(s"Unrecognized GetInfoType value: $infoType")
+ }
+ }
+
override def close(): Unit = {
sessionEvent.endTime = System.currentTimeMillis()
EventBus.post(sessionEvent)
super.close()
- spark.sessionState.catalog.getTempViewNames().foreach(spark.catalog.uncacheTable(_))
+ spark.sessionState.catalog.getTempViewNames().foreach(spark.catalog.uncacheTable)
sessionManager.operationManager.asInstanceOf[SparkSQLOperationManager].closeILoop(handle)
}
}
diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
index d1ab4856f..2ac0235c6 100644
--- a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
+++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
@@ -26,10 +26,12 @@ import org.apache.hadoop.io.Text
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
import org.apache.hadoop.security.token.{Token, TokenIdentifier}
import org.apache.hive.service.rpc.thrift._
+import org.apache.spark.SPARK_VERSION
import org.apache.spark.kyuubi.SparkContextHelper
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.types._
+import org.apache.kyuubi.engine.SemanticVersion
import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
import org.apache.kyuubi.engine.spark.schema.SchemaHelper.TIMESTAMP_NTZ
import org.apache.kyuubi.engine.spark.shim.SparkCatalogShim
@@ -55,6 +57,17 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
}
}
+ test("audit Spark engine MetaData") {
+ withJdbcStatement() { statement =>
+ val metaData = statement.getConnection.getMetaData
+ assert(metaData.getDatabaseProductName === "Spark SQL")
+ assert(metaData.getDatabaseProductVersion === SPARK_VERSION)
+ val ver = SemanticVersion(SPARK_VERSION)
+ assert(metaData.getDatabaseMajorVersion === ver.majorVersion)
+ assert(metaData.getDatabaseMinorVersion === ver.minorVersion)
+ }
+ }
+
test("get columns operation") {
val tableName = "spark_get_col_operation"
var schema = new StructType()
diff --git a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuite.scala b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuite.scala
index fa4fb6ad9..b33669702 100644
--- a/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuite.scala
+++ b/integration-tests/kyuubi-flink-it/src/test/scala/org/apache/kyuubi/it/flink/operation/FlinkOperationSuite.scala
@@ -17,6 +17,8 @@
package org.apache.kyuubi.it.flink.operation
+import org.apache.hive.service.rpc.thrift.{TGetInfoReq, TGetInfoType}
+
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf._
import org.apache.kyuubi.it.flink.WithKyuubiServerAndFlinkMiniCluster
@@ -69,4 +71,26 @@ class FlinkOperationSuite extends WithKyuubiServerAndFlinkMiniCluster
assert(success)
}
}
+
+ test("server info provider - server") {
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "SERVER"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Kyuubi (Incubating)")
+ }
+ }
+ }
+
+ test("server info provider - engine") {
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "ENGINE"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Flink")
+ }
+ }
+ }
}
diff --git a/integration-tests/kyuubi-hive-it/src/test/scala/org/apache/kyuubi/it/hive/operation/KyuubiOperationHiveEnginePerUserSuite.scala b/integration-tests/kyuubi-hive-it/src/test/scala/org/apache/kyuubi/it/hive/operation/KyuubiOperationHiveEnginePerUserSuite.scala
index 613cf185c..de4b9f5cb 100644
--- a/integration-tests/kyuubi-hive-it/src/test/scala/org/apache/kyuubi/it/hive/operation/KyuubiOperationHiveEnginePerUserSuite.scala
+++ b/integration-tests/kyuubi-hive-it/src/test/scala/org/apache/kyuubi/it/hive/operation/KyuubiOperationHiveEnginePerUserSuite.scala
@@ -17,6 +17,8 @@
package org.apache.kyuubi.it.hive.operation
+import org.apache.hive.service.rpc.thrift.{TGetInfoReq, TGetInfoType}
+
import org.apache.kyuubi.{HiveEngineTests, Utils, WithKyuubiServer}
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf._
@@ -37,4 +39,26 @@ class KyuubiOperationHiveEnginePerUserSuite extends WithKyuubiServer with HiveEn
}
override protected def jdbcUrl: String = getJdbcUrl
+
+ test("server info provider - server") {
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "SERVER"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Kyuubi (Incubating)")
+ }
+ }
+ }
+
+ test("server info provider - engine") {
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "ENGINE"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Hive")
+ }
+ }
+ }
}
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
index 62590d627..0c5c4c63f 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala
@@ -1691,6 +1691,16 @@ object KyuubiConf {
.stringConf
.createOptional
+ val SERVER_INFO_PROVIDER: ConfigEntry[String] =
+ buildConf("kyuubi.server.info.provider")
+ .doc("The server information provider name, some clients may rely on this information" +
+ " to check the server compatibilities and functionalities." +
+ " <li>SERVER: Return Kyuubi server information.</li>" +
+ " <li>ENGINE: Return Kyuubi engine information.</li>")
+ .version("1.6.1")
+ .stringConf
+ .createWithDefault("SERVER")
+
val ENGINE_SPARK_SHOW_PROGRESS: ConfigEntry[Boolean] =
buildConf("kyuubi.session.engine.spark.showProgress")
.doc("When true, show the progress bar in the spark engine log.")
diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
index 64d691b27..96a4ee4a6 100644
--- a/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
+++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala
@@ -19,10 +19,10 @@ package org.apache.kyuubi.session
import scala.collection.JavaConverters._
-import org.apache.hive.service.rpc.thrift.{TGetInfoType, TGetInfoValue, TProtocolVersion, TRowSet, TTableSchema}
+import org.apache.hive.service.rpc.thrift._
import org.apache.kyuubi.{KyuubiSQLException, Logging}
-import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf._
import org.apache.kyuubi.config.KyuubiReservedKeys.KYUUBI_CLIENT_IP_KEY
import org.apache.kyuubi.operation.{Operation, OperationHandle}
import org.apache.kyuubi.operation.FetchOrientation.FetchOrientation
@@ -55,7 +55,7 @@ abstract class AbstractSession(
val normalizedConf: Map[String, String] = sessionManager.validateAndNormalizeConf(conf)
- override lazy val name: Option[String] = normalizedConf.get(KyuubiConf.SESSION_NAME.key)
+ override lazy val name: Option[String] = normalizedConf.get(SESSION_NAME.key)
final private val opHandleSet = new java.util.HashSet[OperationHandle]
@@ -108,8 +108,8 @@ abstract class AbstractSession(
override def getInfo(infoType: TGetInfoType): TGetInfoValue = withAcquireRelease() {
infoType match {
- case TGetInfoType.CLI_SERVER_NAME => TGetInfoValue.stringValue("Apache Kyuubi (Incubating)")
- case TGetInfoType.CLI_DBMS_NAME => TGetInfoValue.stringValue("Apache Kyuubi (Incubating)")
+ case TGetInfoType.CLI_SERVER_NAME | TGetInfoType.CLI_DBMS_NAME =>
+ TGetInfoValue.stringValue("Apache Kyuubi (Incubating)")
case TGetInfoType.CLI_DBMS_VER => TGetInfoValue.stringValue(org.apache.kyuubi.KYUUBI_VERSION)
case TGetInfoType.CLI_ODBC_KEYWORDS => TGetInfoValue.stringValue("Unimplemented")
case TGetInfoType.CLI_MAX_COLUMN_NAME_LEN |
diff --git a/kyuubi-common/src/test/scala/org/apache/kyuubi/operation/SparkMetadataTests.scala b/kyuubi-common/src/test/scala/org/apache/kyuubi/operation/SparkMetadataTests.scala
index b0c1ed759..b9f6d5823 100644
--- a/kyuubi-common/src/test/scala/org/apache/kyuubi/operation/SparkMetadataTests.scala
+++ b/kyuubi-common/src/test/scala/org/apache/kyuubi/operation/SparkMetadataTests.scala
@@ -22,7 +22,6 @@ import java.sql.{DatabaseMetaData, ResultSet, SQLException, SQLFeatureNotSupport
import scala.util.Random
import org.apache.kyuubi.{KYUUBI_VERSION, KyuubiSQLException}
-import org.apache.kyuubi.engine.SemanticVersion
import org.apache.kyuubi.operation.meta.ResultSetSchemaConstant._
// For both `in-memory` and `hive` external catalog
@@ -294,7 +293,7 @@ trait SparkMetadataTests extends HiveJDBCTestHelper {
}
}
- test("audit Kyuubi Hive JDBC connection MetaData") {
+ test("audit Kyuubi Hive JDBC connection common MetaData") {
withJdbcStatement() { statement =>
val metaData = statement.getConnection.getMetaData
Seq(
@@ -405,13 +404,8 @@ trait SparkMetadataTests extends HiveJDBCTestHelper {
assert(metaData.allTablesAreSelectable)
assert(metaData.getClientInfoProperties.next)
- assert(metaData.getDatabaseProductName === "Apache Kyuubi (Incubating)")
- assert(metaData.getDatabaseProductVersion === KYUUBI_VERSION)
assert(metaData.getDriverName === "Kyuubi Project Hive JDBC Shaded Client")
assert(metaData.getDriverVersion === KYUUBI_VERSION)
- val ver = SemanticVersion(KYUUBI_VERSION)
- assert(metaData.getDatabaseMajorVersion === ver.majorVersion)
- assert(metaData.getDatabaseMinorVersion === ver.minorVersion)
assert(
metaData.getIdentifierQuoteString === " ",
"This method returns a space \" \" if identifier quoting is not supported")
@@ -455,21 +449,16 @@ trait SparkMetadataTests extends HiveJDBCTestHelper {
assert(metaData.getDefaultTransactionIsolation === java.sql.Connection.TRANSACTION_NONE)
assert(!metaData.supportsTransactions)
assert(!metaData.getProcedureColumns("", "%", "%", "%").next())
- try {
- assert(!metaData.getPrimaryKeys("", "default", "src").next())
- } catch {
- case e: Exception =>
- assert(e.isInstanceOf[SQLException])
- assert(e.getMessage.contains(KyuubiSQLException.featureNotSupported().getMessage))
+ val e1 = intercept[SQLException] {
+ metaData.getPrimaryKeys("", "default", "src").next()
}
+ assert(e1.getMessage.contains(KyuubiSQLException.featureNotSupported().getMessage))
assert(!metaData.getImportedKeys("", "default", "").next())
- try {
- assert(!metaData.getCrossReference("", "default", "src", "", "default", "src2").next())
- } catch {
- case e: Exception =>
- assert(e.isInstanceOf[SQLException])
- assert(e.getMessage.contains(KyuubiSQLException.featureNotSupported().getMessage))
+
+ val e2 = intercept[SQLException] {
+ metaData.getCrossReference("", "default", "src", "", "default", "src2").next()
}
+ assert(e2.getMessage.contains(KyuubiSQLException.featureNotSupported().getMessage))
assert(!metaData.getIndexInfo("", "default", "src", true, true).next())
assert(metaData.supportsResultSetType(new Random().nextInt()))
diff --git a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/TFrontendServiceSuite.scala b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/TFrontendServiceSuite.scala
index 1e730abc6..9972c4f60 100644
--- a/kyuubi-common/src/test/scala/org/apache/kyuubi/service/TFrontendServiceSuite.scala
+++ b/kyuubi-common/src/test/scala/org/apache/kyuubi/service/TFrontendServiceSuite.scala
@@ -26,7 +26,7 @@ import org.scalatest.time._
import org.apache.kyuubi.{KyuubiFunSuite, KyuubiSQLException, Utils}
import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.config.KyuubiConf.{FRONTEND_BIND_HOST, FRONTEND_CONNECTION_URL_USE_HOSTNAME, FRONTEND_THRIFT_BINARY_BIND_HOST, FRONTEND_THRIFT_BINARY_BIND_PORT}
+import org.apache.kyuubi.config.KyuubiConf._
import org.apache.kyuubi.operation.{OperationHandle, TClientTestUtils}
import org.apache.kyuubi.service.TFrontendService.FeServiceServerContext
import org.apache.kyuubi.session.{AbstractSession, SessionHandle}
@@ -34,7 +34,7 @@ import org.apache.kyuubi.session.{AbstractSession, SessionHandle}
class TFrontendServiceSuite extends KyuubiFunSuite {
protected val server = new NoopTBinaryFrontendServer()
- protected val conf = KyuubiConf()
+ protected val conf: KyuubiConf = KyuubiConf()
.set(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
.set("kyuubi.test.server.should.fail", "false")
.set(KyuubiConf.SESSION_CHECK_INTERVAL, Duration.ofSeconds(5).toMillis)
@@ -54,8 +54,8 @@ class TFrontendServiceSuite extends KyuubiFunSuite {
}
override def afterAll(): Unit = {
- server.getServices.foreach(_.stop())
super.afterAll()
+ server.getServices.foreach(_.stop())
}
private def checkOperationResult(
@@ -166,8 +166,7 @@ class TFrontendServiceSuite extends KyuubiFunSuite {
val req = new TGetInfoReq()
req.setSessionHandle(handle)
req.setInfoType(TGetInfoType.CLI_DBMS_VER)
- val resp = client.GetInfo(req)
- assert(resp.getInfoValue.getStringValue === org.apache.kyuubi.KYUUBI_VERSION)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === org.apache.kyuubi.KYUUBI_VERSION)
req.setInfoType(TGetInfoType.CLI_SERVER_NAME)
assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Kyuubi (Incubating)")
req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
index 1b1694e15..edbb5c45e 100644
--- a/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
+++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
@@ -199,12 +199,13 @@ class KyuubiSessionImpl(
}
override def getInfo(infoType: TGetInfoType): TGetInfoValue = {
- if (client != null) {
- withAcquireRelease() {
- client.getInfo(infoType).getInfoValue
- }
- } else {
- super.getInfo(infoType)
+ sessionConf.get(SERVER_INFO_PROVIDER) match {
+ case "SERVER" => super.getInfo(infoType)
+ case "ENGINE" => withAcquireRelease() {
+ waitForEngineLaunched()
+ client.getInfo(infoType).getInfoValue
+ }
+ case unknown => throw new IllegalArgumentException(s"Unknown server info provider $unknown")
}
}
}
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
index 4f78c7f42..db23bf8c0 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerConnectionSuite.scala
@@ -214,7 +214,9 @@ class KyuubiOperationPerConnectionSuite extends WithKyuubiServer with HiveJDBCTe
}
test("transfer the TGetInfoReq to kyuubi engine side to verify the connection valid") {
- withSessionConf(Map.empty)(Map(KyuubiConf.SESSION_ENGINE_LAUNCH_ASYNC.key -> "false"))() {
+ withSessionConf(Map.empty)(Map(
+ KyuubiConf.SERVER_INFO_PROVIDER.key -> "ENGINE",
+ KyuubiConf.SESSION_ENGINE_LAUNCH_ASYNC.key -> "false"))() {
withJdbcStatement() { statement =>
val conn = statement.getConnection.asInstanceOf[KyuubiConnection]
assert(conn.isValid(3000))
diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
index c2f1be1d4..0cb38a1cc 100644
--- a/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
+++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/KyuubiOperationPerUserSuite.scala
@@ -20,12 +20,13 @@ package org.apache.kyuubi.operation
import java.util.UUID
import org.apache.hadoop.fs.{FileSystem, FileUtil, Path}
-import org.apache.hive.service.rpc.thrift.{TExecuteStatementReq, TStatusCode}
+import org.apache.hive.service.rpc.thrift.{TExecuteStatementReq, TGetInfoReq, TGetInfoType, TStatusCode}
import org.scalatest.time.SpanSugar._
-import org.apache.kyuubi.{Utils, WithKyuubiServer, WithSimpleDFSService}
+import org.apache.kyuubi.{KYUUBI_VERSION, Utils, WithKyuubiServer, WithSimpleDFSService}
import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.config.KyuubiConf.KYUUBI_ENGINE_ENV_PREFIX
+import org.apache.kyuubi.engine.SemanticVersion
import org.apache.kyuubi.jdbc.hive.KyuubiStatement
import org.apache.kyuubi.session.{KyuubiSessionImpl, KyuubiSessionManager, SessionHandle}
@@ -43,6 +44,17 @@ class KyuubiOperationPerUserSuite
conf.set(s"$KYUUBI_ENGINE_ENV_PREFIX.HADOOP_CONF_DIR", getHadoopConfDir)
}
+ test("audit Kyuubi server MetaData") {
+ withJdbcStatement() { statement =>
+ val metaData = statement.getConnection.getMetaData
+ assert(metaData.getDatabaseProductName === "Apache Kyuubi (Incubating)")
+ assert(metaData.getDatabaseProductVersion === KYUUBI_VERSION)
+ val ver = SemanticVersion(KYUUBI_VERSION)
+ assert(metaData.getDatabaseMajorVersion === ver.majorVersion)
+ assert(metaData.getDatabaseMinorVersion === ver.minorVersion)
+ }
+ }
+
test("kyuubi defined function - system_user/session_user") {
withJdbcStatement() { statement =>
val rs = statement.executeQuery("SELECT system_user(), session_user()")
@@ -168,46 +180,45 @@ class KyuubiOperationPerUserSuite
}
test("support to interrupt the thrift request if remote engine is broken") {
- if (!httpMode) {
- withSessionConf(Map(
- KyuubiConf.ENGINE_ALIVE_PROBE_ENABLED.key -> "true",
- KyuubiConf.ENGINE_ALIVE_PROBE_INTERVAL.key -> "1000",
- KyuubiConf.ENGINE_ALIVE_TIMEOUT.key -> "1000"))(Map.empty)(
- Map.empty) {
- withSessionHandle { (client, handle) =>
- val preReq = new TExecuteStatementReq()
- preReq.setStatement("select engine_name()")
- preReq.setSessionHandle(handle)
- preReq.setRunAsync(false)
- client.ExecuteStatement(preReq)
-
- val sessionHandle = SessionHandle(handle)
- val session = server.backendService.sessionManager.asInstanceOf[KyuubiSessionManager]
- .getSession(sessionHandle).asInstanceOf[KyuubiSessionImpl]
- session.client.getEngineAliveProbeProtocol.foreach(_.getTransport.close())
-
- val exitReq = new TExecuteStatementReq()
- exitReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 1000L)," +
- "java_method('java.lang.System', 'exit', 1)")
- exitReq.setSessionHandle(handle)
- exitReq.setRunAsync(true)
- client.ExecuteStatement(exitReq)
-
- val executeStmtReq = new TExecuteStatementReq()
- executeStmtReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 30000l)")
- executeStmtReq.setSessionHandle(handle)
- executeStmtReq.setRunAsync(false)
- val startTime = System.currentTimeMillis()
- val executeStmtResp = client.ExecuteStatement(executeStmtReq)
- assert(executeStmtResp.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
- assert(executeStmtResp.getStatus.getErrorMessage.contains(
- "java.net.SocketException: Connection reset") ||
- executeStmtResp.getStatus.getErrorMessage.contains(
- "Caused by: java.net.SocketException: Broken pipe (Write failed)"))
- val elapsedTime = System.currentTimeMillis() - startTime
- assert(elapsedTime < 20 * 1000)
- assert(session.client.asyncRequestInterrupted)
- }
+ assume(!httpMode)
+ withSessionConf(Map(
+ KyuubiConf.ENGINE_ALIVE_PROBE_ENABLED.key -> "true",
+ KyuubiConf.ENGINE_ALIVE_PROBE_INTERVAL.key -> "1000",
+ KyuubiConf.ENGINE_ALIVE_TIMEOUT.key -> "1000"))(Map.empty)(
+ Map.empty) {
+ withSessionHandle { (client, handle) =>
+ val preReq = new TExecuteStatementReq()
+ preReq.setStatement("select engine_name()")
+ preReq.setSessionHandle(handle)
+ preReq.setRunAsync(false)
+ client.ExecuteStatement(preReq)
+
+ val sessionHandle = SessionHandle(handle)
+ val session = server.backendService.sessionManager.asInstanceOf[KyuubiSessionManager]
+ .getSession(sessionHandle).asInstanceOf[KyuubiSessionImpl]
+ session.client.getEngineAliveProbeProtocol.foreach(_.getTransport.close())
+
+ val exitReq = new TExecuteStatementReq()
+ exitReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 1000L)," +
+ "java_method('java.lang.System', 'exit', 1)")
+ exitReq.setSessionHandle(handle)
+ exitReq.setRunAsync(true)
+ client.ExecuteStatement(exitReq)
+
+ val executeStmtReq = new TExecuteStatementReq()
+ executeStmtReq.setStatement("SELECT java_method('java.lang.Thread', 'sleep', 30000l)")
+ executeStmtReq.setSessionHandle(handle)
+ executeStmtReq.setRunAsync(false)
+ val startTime = System.currentTimeMillis()
+ val executeStmtResp = client.ExecuteStatement(executeStmtReq)
+ assert(executeStmtResp.getStatus.getStatusCode === TStatusCode.ERROR_STATUS)
+ assert(executeStmtResp.getStatus.getErrorMessage.contains(
+ "java.net.SocketException: Connection reset") ||
+ executeStmtResp.getStatus.getErrorMessage.contains(
+ "Caused by: java.net.SocketException: Broken pipe (Write failed)"))
+ val elapsedTime = System.currentTimeMillis() - startTime
+ assert(elapsedTime < 20 * 1000)
+ assert(session.client.asyncRequestInterrupted)
}
}
}
@@ -243,4 +254,28 @@ class KyuubiOperationPerUserSuite
assert(rs.getString(1) === "3")
}
}
+
+ test("server info provider - server") {
+ assume(!httpMode)
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "SERVER"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Apache Kyuubi (Incubating)")
+ }
+ }
+ }
+
+ test("server info provider - engine") {
+ assume(!httpMode)
+ withSessionConf(Map(KyuubiConf.SERVER_INFO_PROVIDER.key -> "ENGINE"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TGetInfoReq()
+ req.setSessionHandle(handle)
+ req.setInfoType(TGetInfoType.CLI_DBMS_NAME)
+ assert(client.GetInfo(req).getInfoValue.getStringValue === "Spark SQL")
+ }
+ }
+ }
}