This is an automated email from the ASF dual-hosted git repository. zhouyuan pushed a commit to branch revert-12823-iceberg-call-parser in repository https://gitbox.apache.org/repos/asf/gluten.git
commit 2a92e0c064360bc7d82573bf6317380f1961f34e Author: Yuan <[email protected]> AuthorDate: Thu Aug 27 12:19:49 2026 +0100 Revert "[VL][Iceberg] Register iceberg spark session extensions automatically…" This reverts commit 0364eaf6e05ea642b2a643bc9b5b1757159ebc12. --- .../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, 13 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/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]
