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 e0dce586b0 [VL][UT] Add the iceberg jar when the iceberg profile is
selected (#12914)
e0dce586b0 is described below
commit e0dce586b0a208c67eb313e0b9805515d3fbda73
Author: inf <[email protected]>
AuthorDate: Mon Aug 31 12:12:26 2026 +0000
[VL][UT] Add the iceberg jar when the iceberg profile is selected (#12914)
---
.../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-ut/pom.xml | 14 ++++++
.../gluten/GlutenSessionExtensionsSuite.scala | 52 ----------------------
6 files changed, 27 insertions(+), 85 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 71414a7be4..8b6af1f045 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,19 +23,15 @@ 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(icebergSparkSessionExtension)
+ SparkReflectionUtil.isClassPresent(
+ "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
}
- 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 05a1aff350..81fea4727b 100644
--- a/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
+++ b/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala
@@ -50,11 +50,17 @@ private[gluten] class GlutenDriverPlugin extends
DriverPlugin with Logging {
override def init(sc: SparkContext, pluginContext: PluginContext):
util.Map[String, String] = {
val conf = pluginContext.conf()
- val components = Component.sorted()
- configureSessionExtensions(conf, components)
+ // 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)
+ }
setPredefinedConfigs(conf)
+ val components = Component.sorted()
printComponentInfo(components)
setComponentInfoConfig(conf, components)
components.foreach(_.onDriverStart(sc, pluginContext))
@@ -71,19 +77,6 @@ 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 f5e6784536..e2256c01e0 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,8 +79,6 @@ 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 b5ea553d17..56f3fbdace 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,21 +37,14 @@ 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/pom.xml b/gluten-ut/pom.xml
index 1316d878c7..a075ba35dc 100644
--- a/gluten-ut/pom.xml
+++ b/gluten-ut/pom.xml
@@ -200,6 +200,20 @@
</build>
<profiles>
+ <profile>
+ <id>iceberg</id>
+ <activation>
+ <activeByDefault>false</activeByDefault>
+ </activation>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.iceberg</groupId>
+
<artifactId>iceberg-spark-runtime-${sparkbundle.version}_${scala.binary.version}</artifactId>
+ <version>${iceberg.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </profile>
<profile>
<id>spark-3.3</id>
<modules>
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
deleted file mode 100644
index 4165cac244..0000000000
---
a/gluten-ut/test/src/test/scala/org/apache/gluten/GlutenSessionExtensionsSuite.scala
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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]
