From 2a92e0c064360bc7d82573bf6317380f1961f34e Mon Sep 17 00:00:00 2001 From: Yuan Date: Thu, 27 Aug 2026 12:19:49 +0100 Subject: [PATCH 1/2] =?UTF-8?q?Revert=20"[VL][Iceberg]=20Register=20iceber?= =?UTF-8?q?g=20spark=20session=20extensions=20automatically=E2=80=A6"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 0364eaf6e05ea642b2a643bc9b5b1757159ebc12. --- .../component/VeloxIcebergComponent.scala | 8 +-- .../org/apache/gluten/GlutenPlugin.scala | 23 +++----- .../apache/gluten/component/Component.scala | 2 - .../gluten/execution/IcebergSuite.scala | 13 ++--- .../gluten/GlutenSessionExtensionsSuite.scala | 52 ------------------- 5 files changed, 13 insertions(+), 85 deletions(-) delete mode 100644 gluten-ut/test/src/test/scala/org/apache/gluten/GlutenSessionExtensionsSuite.scala 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 71414a7be4e..8b6af1f045d 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 05a1aff350c..81fea4727b8 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 f5e6784536d..e2256c01e0c 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 b5ea553d17f..56f3fbdace7 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/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 4165cac244d..00000000000 --- 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 = {} - } -} From d2e5a3def7c21ab118f63540574bcc95f1ade191 Mon Sep 17 00:00:00 2001 From: Hazmi Date: Thu, 27 Aug 2026 11:34:39 +0000 Subject: [PATCH 2/2] [VL][UT] Add the iceberg jar when the iceberg profile is selected --- gluten-ut/pom.xml | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/gluten-ut/pom.xml b/gluten-ut/pom.xml index 1316d878c78..a075ba35dcf 100644 --- a/gluten-ut/pom.xml +++ b/gluten-ut/pom.xml @@ -200,6 +200,20 @@ + + iceberg + + false + + + + org.apache.iceberg + iceberg-spark-runtime-${sparkbundle.version}_${scala.binary.version} + ${iceberg.version} + test + + + spark-3.3