This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 0364eaf6e0 [VL][Iceberg] Register iceberg spark session extensions
automatically (#12823)
0364eaf6e0 is described below
commit 0364eaf6e05ea642b2a643bc9b5b1757159ebc12
Author: inf <[email protected]>
AuthorDate: Mon Aug 24 11:08:55 2026 +0000
[VL][Iceberg] Register iceberg spark session extensions automatically
(#12823)
---
.../gluten/component/VeloxIcebergComponent.scala | 8 +++-
.../scala/org/apache/gluten/GlutenPlugin.scala | 23 ++++++----
.../org/apache/gluten/component/Component.scala | 2 +
.../org/apache/gluten/execution/IcebergSuite.scala | 13 ++++--
.../gluten/GlutenSessionExtensionsSuite.scala | 52 ++++++++++++++++++++++
5 files changed, 85 insertions(+), 13 deletions(-)
diff --git
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/component/VeloxIcebergComponent.scala
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/component/VeloxIcebergComponent.scala
index 8b6af1f045..71414a7be4 100644
---
a/backends-velox/src-iceberg/main/scala/org/apache/gluten/component/VeloxIcebergComponent.scala
+++
b/backends-velox/src-iceberg/main/scala/org/apache/gluten/component/VeloxIcebergComponent.scala
@@ -23,15 +23,19 @@ import org.apache.gluten.extension.injector.Injector
import org.apache.spark.util.SparkReflectionUtil
class VeloxIcebergComponent extends Component {
+ private val icebergSparkSessionExtension =
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions"
+
override def name(): String = "velox-iceberg"
override def dependencies(): Seq[Class[_ <: Component]] =
classOf[VeloxBackend] :: Nil
override def isRuntimeCompatible: Boolean = {
- SparkReflectionUtil.isClassPresent(
- "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
+ SparkReflectionUtil.isClassPresent(icebergSparkSessionExtension)
}
+ override def sparkSessionExtensions(): Seq[String] =
icebergSparkSessionExtension :: Nil
+
override def injectRules(injector: Injector): Unit = {
OffloadIcebergScan.inject(injector)
OffloadIcebergWrite.inject(injector)
diff --git a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
index 81fea4727b..05a1aff350 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
@@ -50,17 +50,11 @@ private[gluten] class GlutenDriverPlugin extends
DriverPlugin with Logging {
override def init(sc: SparkContext, pluginContext: PluginContext):
util.Map[String, String] = {
val conf = pluginContext.conf()
- // Spark SQL extensions
- val extensionSeq = conf.get(SPARK_SESSION_EXTENSIONS).getOrElse(Seq.empty)
- if
(!extensionSeq.toSet.contains(GlutenSessionExtensions.GLUTEN_SESSION_EXTENSION_NAME))
{
- conf.set(
- SPARK_SESSION_EXTENSIONS,
- extensionSeq :+ GlutenSessionExtensions.GLUTEN_SESSION_EXTENSION_NAME)
- }
+ val components = Component.sorted()
+ configureSessionExtensions(conf, components)
setPredefinedConfigs(conf)
- val components = Component.sorted()
printComponentInfo(components)
setComponentInfoConfig(conf, components)
components.foreach(_.onDriverStart(sc, pluginContext))
@@ -77,6 +71,19 @@ private[gluten] class GlutenDriverPlugin extends
DriverPlugin with Logging {
}
private object GlutenDriverPlugin extends Logging {
+ private[gluten] def configureSessionExtensions(
+ conf: SparkConf,
+ components: Seq[Component]): Unit = {
+ val configuredExtensions =
conf.get(SPARK_SESSION_EXTENSIONS).getOrElse(Seq.empty)
+ val requiredExtensions =
+ components.flatMap(_.sparkSessionExtensions()) :+
+ GlutenSessionExtensions.GLUTEN_SESSION_EXTENSION_NAME
+ val mergedExtensions = (configuredExtensions ++
requiredExtensions).distinct
+ if (mergedExtensions != configuredExtensions) {
+ conf.set(SPARK_SESSION_EXTENSIONS, mergedExtensions)
+ }
+ }
+
private def checkOffHeapSettings(conf: SparkConf): Unit = {
if (conf.get(GlutenCoreConfig.DYNAMIC_OFFHEAP_SIZING_ENABLED)) {
// When dynamic off-heap sizing is enabled, off-heap mode is not
strictly required to be
diff --git
a/gluten-core/src/main/scala/org/apache/gluten/component/Component.scala
b/gluten-core/src/main/scala/org/apache/gluten/component/Component.scala
index e2256c01e0..f5e6784536 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/component/Component.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/component/Component.scala
@@ -79,6 +79,8 @@ trait Component {
def info(): Map[String, String] = Map.empty
def dependencies(): Seq[Class[_ <: Component]]
+ def sparkSessionExtensions(): Seq[String] = Nil
+
/** Spark listeners. */
def onDriverStart(sc: SparkContext, pc: PluginContext): Unit = {}
def onDriverShutdown(): Unit = {}
diff --git
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
index 56f3fbdace..b5ea553d17 100644
---
a/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
+++
b/gluten-iceberg/src/test/scala/org/apache/gluten/execution/IcebergSuite.scala
@@ -37,14 +37,21 @@ abstract class IcebergSuite extends
WholeStageTransformerSuite {
.set("spark.memory.offHeap.size", "2g")
.set("spark.unsafe.exceptionOnMemoryLeak", "true")
.set("spark.sql.autoBroadcastJoinThreshold", "-1")
- .set(
- "spark.sql.extensions",
- "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.set("spark.sql.catalog.spark_catalog",
"org.apache.iceberg.spark.SparkCatalog")
.set("spark.sql.catalog.spark_catalog.type", "hadoop")
.set("spark.sql.catalog.spark_catalog.warehouse",
s"file://$rootPath/tpch-data-iceberg-velox")
}
+ test("iceberg system procedures are registered by the Gluten plugin") {
+ spark.sessionState.sqlParser.parsePlan(
+ """
+ |CALL spark_catalog.system.register_table(
+ | table => 'default.issue_12693',
+ | metadata_file => 'file:///tmp/does-not-exist.metadata.json'
+ |)
+ |""".stripMargin)
+ }
+
test("iceberg transformer exists") {
withTable("iceberg_tb") {
spark.sql("""
diff --git
a/gluten-ut/test/src/test/scala/org/apache/gluten/GlutenSessionExtensionsSuite.scala
b/gluten-ut/test/src/test/scala/org/apache/gluten/GlutenSessionExtensionsSuite.scala
new file mode 100644
index 0000000000..4165cac244
--- /dev/null
+++
b/gluten-ut/test/src/test/scala/org/apache/gluten/GlutenSessionExtensionsSuite.scala
@@ -0,0 +1,52 @@
+/*
+ * 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.gluten
+
+import org.apache.gluten.component.Component
+import org.apache.gluten.extension.GlutenSessionExtensions
+import org.apache.gluten.extension.injector.Injector
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.internal.StaticSQLConf.SPARK_SESSION_EXTENSIONS
+
+import org.scalatest.funsuite.AnyFunSuite
+
+class GlutenSessionExtensionsSuite extends AnyFunSuite {
+ test("component session extensions are appended once") {
+ val configuredExtension = "example.ConfiguredExtension"
+ val componentExtension = "example.ComponentExtension"
+ val conf = new SparkConf(false).set(
+ SPARK_SESSION_EXTENSIONS.key,
+ Seq(configuredExtension, componentExtension).mkString(","))
+ val component = new TestComponent(Seq(componentExtension,
componentExtension))
+
+ GlutenDriverPlugin.configureSessionExtensions(conf, component :: Nil)
+
+ assert(
+ conf.get(SPARK_SESSION_EXTENSIONS.key).split(",").toSeq == Seq(
+ configuredExtension,
+ componentExtension,
+ GlutenSessionExtensions.GLUTEN_SESSION_EXTENSION_NAME))
+ }
+
+ private class TestComponent(extensions: Seq[String]) extends Component {
+ override def name(): String = "test"
+ override def dependencies(): Seq[Class[_ <: Component]] = Nil
+ override def sparkSessionExtensions(): Seq[String] = extensions
+ override def injectRules(injector: Injector): Unit = {}
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
