You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by ul...@apache.org on 2022/09/15 05:18:47 UTC
[incubator-kyuubi] branch master updated: [KYUUBI #3464] Support for pooling external catalog
This is an automated email from the ASF dual-hosted git repository.
ulyssesyou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new c3c770720 [KYUUBI #3464] Support for pooling external catalog
c3c770720 is described below
commit c3c77072033bffd0f6ba9fdf6bf10e548e5e1625
Author: yikf <yi...@gmail.com>
AuthorDate: Thu Sep 15 13:18:28 2022 +0800
[KYUUBI #3464] Support for pooling external catalog
### _Why are the changes needed?_
Fix https://github.com/apache/incubator-kyuubi/issues/3464, currently, Kyuubi supports hive connector for read/write hive table, it is implemented based on the [Apache Spark DataSource V2](https://www.databricks.com/session/apache-spark-data-source-v2), but there's a potential issue;
Kyuubi use `kyuubi.engine.single.spark.session`=[false](https://kyuubi.apache.org/docs/latest/deployment/settings.html#:~:text=kyuubi.engine.single.spark.session) to provide concurrency sql execution in context isolation, this cause spark.newSession invoked for each transaction, in spark v1 catalog, externalCatalog is shared in the mutiple session, but in catalog v2 architecture, it's big different with v1, v2 catalogs are managed by `CatalogManager` which is [session level](https://g [...]
This issue aims to pool externalCatalog to address the potential issues mentioned above.
### _How was this patch tested?_
- [ ] 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.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #3465 from Yikf/catalog-pool.
Closes #3464
5e8a94dd [yikf] Support for pooling external catalog
Authored-by: yikf <yi...@gmail.com>
Signed-off-by: ulysses-you <ul...@apache.org>
---
.../connector/hive/ExternalCatalogManager.scala | 90 ++++++++++++++++++++++
.../hive/ExternalCatalogSharePolicy.scala | 53 +++++++++++++
.../spark/connector/hive/HiveTableCatalog.scala | 6 +-
.../connector/hive/KyuubiHiveConnectorConf.scala | 42 ++++++++++
.../connector/hive/ExternalCatalogPoolSuite.scala | 68 ++++++++++++++++
5 files changed, 257 insertions(+), 2 deletions(-)
diff --git a/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogManager.scala b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogManager.scala
new file mode 100644
index 000000000..6de4f6d78
--- /dev/null
+++ b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogManager.scala
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive
+
+import java.util.concurrent.ConcurrentHashMap
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper.HiveExternalCatalog
+
+import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.EXTERNAL_CATALOG_SHARE_POLICY
+
+object ExternalCatalogManager {
+ private var manager: ExternalCatalogManager = _
+
+ def getOrCreate(sparkSession: SparkSession): ExternalCatalogManager = {
+ if (manager == null) {
+ synchronized {
+ if (manager == null) {
+ val conf = sparkSession.sessionState.conf
+ manager = ExternalCatalogSharePolicy.fromString(
+ conf.getConf(EXTERNAL_CATALOG_SHARE_POLICY)) match {
+ case OneForAllPolicy => new OneForAllPolicyManager()
+ case OneForOnePolicy => OneForOnePolicyManager
+ }
+ }
+ }
+ }
+ manager
+ }
+
+ private[kyuubi] def reset(): Unit = {
+ if (manager != null) {
+ manager.invalidateAll()
+ manager = null
+ }
+ }
+}
+
+abstract class ExternalCatalogManager {
+
+ def take(ticket: Ticket): HiveExternalCatalog
+
+ def invalidateAll(): Unit = {}
+}
+
+/**
+ * A [[OneForAllPolicy]] policy for the externalCatalog manager, which caches the externalCatalog
+ * according to the catalogName, aiming for only one of each catalog globally.
+ */
+class OneForAllPolicyManager() extends ExternalCatalogManager {
+ private val catalogCache = new ConcurrentHashMap[String, HiveExternalCatalog]()
+
+ override def take(ticket: Ticket): HiveExternalCatalog = {
+ catalogCache.computeIfAbsent(
+ ticket.catalogName,
+ _ => {
+ new HiveExternalCatalog(ticket.sparkConf, ticket.hadoopConf)
+ })
+ }
+}
+
+/**
+ * A [[OneForOnePolicy]] policy for the externalCatalog manager, It doesn't actually cache any
+ * externalCatalog, each session will have its own externalCatalog.
+ */
+object OneForOnePolicyManager extends ExternalCatalogManager {
+
+ override def take(ticket: Ticket): HiveExternalCatalog = {
+ new HiveExternalCatalog(ticket.sparkConf, ticket.hadoopConf)
+ }
+}
+
+case class Ticket(catalogName: String, sparkConf: SparkConf, hadoopConf: Configuration)
diff --git a/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogSharePolicy.scala b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogSharePolicy.scala
new file mode 100644
index 000000000..37edc3d04
--- /dev/null
+++ b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogSharePolicy.scala
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive
+
+import java.util.Locale
+
+sealed trait ExternalCatalogSharePolicy {
+
+ /**
+ * String name of the share policy
+ */
+ def name: String
+}
+
+/**
+ * Indicate to an external catalog is shared globally with the HiveCatalogs
+ * with the same catalogName.
+ */
+case object OneForAllPolicy extends ExternalCatalogSharePolicy { val name = "ONE_FOR_ALL" }
+
+/**
+ * Indicate to an external catalog is used by only one HiveCatalog.
+ */
+case object OneForOnePolicy extends ExternalCatalogSharePolicy { val name = "ONE_FOR_ONE" }
+
+object ExternalCatalogSharePolicy {
+
+ /**
+ * Returns the share policy from the given string.
+ */
+ def fromString(policy: String): ExternalCatalogSharePolicy =
+ policy.toUpperCase(Locale.ROOT) match {
+ case OneForAllPolicy.name => OneForAllPolicy
+ case OneForOnePolicy.name => OneForOnePolicy
+ case _ => throw new IllegalArgumentException(s"Unknown share policy: $policy. Accepted " +
+ "policies are 'ONE_FOR_ONE', 'ONE_FOR_ALL'.")
+ }
+}
diff --git a/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala
index ef36392f1..e58a2d871 100644
--- a/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala
+++ b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/HiveTableCatalog.scala
@@ -38,7 +38,7 @@ import org.apache.spark.sql.connector.catalog.NamespaceChange.RemoveProperty
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.execution.datasources.DataSource
import org.apache.spark.sql.hive.HiveUDFExpressionBuilder
-import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper.{catalogV2Util, postExternalCatalogEvent, HiveExternalCatalog, HiveMetastoreCatalog, HiveSessionCatalog}
+import org.apache.spark.sql.hive.kyuubi.connector.HiveBridgeHelper.{catalogV2Util, postExternalCatalogEvent, HiveMetastoreCatalog, HiveSessionCatalog}
import org.apache.spark.sql.internal.StaticSQLConf.CATALOG_IMPLEMENTATION
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -53,6 +53,8 @@ class HiveTableCatalog(sparkSession: SparkSession)
def this() = this(SparkSession.active)
+ private val externalCatalogManager = ExternalCatalogManager.getOrCreate(sparkSession)
+
private val sc = sparkSession.sparkContext
private val sessionState = sparkSession.sessionState
@@ -116,7 +118,7 @@ class HiveTableCatalog(sparkSession: SparkSession)
* A catalog that interacts with external systems.
*/
lazy val externalCatalog: ExternalCatalogWithListener = {
- val externalCatalog = new HiveExternalCatalog(sparkConf, hadoopConf)
+ val externalCatalog = externalCatalogManager.take(Ticket(catalogName, sparkConf, hadoopConf))
// Wrap to provide catalog events
val wrapped = new ExternalCatalogWithListener(externalCatalog)
diff --git a/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/KyuubiHiveConnectorConf.scala b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/KyuubiHiveConnectorConf.scala
new file mode 100644
index 000000000..cc5ffde9c
--- /dev/null
+++ b/extensions/spark/kyuubi-spark-connector-hive/src/main/scala/org/apache/kyuubi/spark/connector/hive/KyuubiHiveConnectorConf.scala
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive
+
+import java.util.Locale
+
+object KyuubiHiveConnectorConf {
+
+ import org.apache.spark.sql.internal.SQLConf.buildStaticConf
+
+ val EXTERNAL_CATALOG_SHARE_POLICY =
+ buildStaticConf("spark.sql.kyuubi.hive.connector.externalCatalog.share.policy")
+ .internal()
+ .doc(s"Indicates the share policy for the externalCatalog in the Kyuubi Connector, we use" +
+ "'all' by default. " +
+ "<li>ONE_FOR_ONE: Indicate to an external catalog is used by only one HiveCatalog. </li> " +
+ "<li>ONE_FOR_ALL: Indicate to an external catalog is shared globally with the " +
+ "HiveCatalogs with the same catalogName. </li> ")
+ .version("1.7.0")
+ .stringConf
+ .transform(policy => policy.toUpperCase(Locale.ROOT))
+ .checkValue(
+ policy => Set("ONE_FOR_ONE", "ONE_FOR_ALL").contains(policy),
+ "Invalid value for 'spark.sql.kyuubi.hive.connector.externalCatalog.share.policy'." +
+ "Valid values are 'ONE_FOR_ONE', 'ONE_FOR_ALL'.")
+ .createWithDefault(OneForAllPolicy.name)
+}
diff --git a/extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogPoolSuite.scala b/extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogPoolSuite.scala
new file mode 100644
index 000000000..7c02e8531
--- /dev/null
+++ b/extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/ExternalCatalogPoolSuite.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kyuubi.spark.connector.hive
+
+import org.apache.kyuubi.spark.connector.hive.KyuubiHiveConnectorConf.EXTERNAL_CATALOG_SHARE_POLICY
+
+class ExternalCatalogPoolSuite extends KyuubiHiveTest {
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ ExternalCatalogManager.reset()
+ }
+
+ test("test configuration for externalCatalog share policy") {
+ withSparkSession() { spark =>
+ val pool1 = ExternalCatalogManager.getOrCreate(spark)
+ val pool2 = ExternalCatalogManager.getOrCreate(spark)
+ assert(pool1.isInstanceOf[OneForAllPolicyManager])
+ assert(pool1 == pool2)
+ }
+
+ ExternalCatalogManager.reset()
+ withSparkSession(Map(EXTERNAL_CATALOG_SHARE_POLICY.key -> "ONE_FOR_ONE")) { spark =>
+ val pool1 = ExternalCatalogManager.getOrCreate(spark)
+ val pool2 = ExternalCatalogManager.getOrCreate(spark)
+ assert(pool1.isInstanceOf[OneForOnePolicyManager.type])
+ assert(pool1 == pool2)
+ }
+ }
+
+ test("ALL policy: external catalog is shared globally " +
+ "with the HiveCatalogs with the same catalogName") {
+ withSparkSession(Map(
+ EXTERNAL_CATALOG_SHARE_POLICY.key -> "ONE_FOR_ALL")) { spark =>
+ val catalog1 =
+ Ticket("catalog1", spark.sparkContext.getConf, spark.sessionState.newHadoopConf())
+ val catalog2 =
+ Ticket("catalog2", spark.sparkContext.getConf, spark.sessionState.newHadoopConf())
+ val pool = ExternalCatalogManager.getOrCreate(spark)
+ val externalCatalog1 = pool.take(catalog1)
+ val externalCatalog2 = pool.take(catalog2)
+
+ assert(externalCatalog1 != externalCatalog2)
+ (1 to 10).foreach { id =>
+ assert(pool.take(catalog1) == externalCatalog1)
+ }
+
+ (1 to 10).foreach { id =>
+ assert(pool.take(catalog2) == externalCatalog2)
+ }
+ }
+ }
+}