This is an automated email from the ASF dual-hosted git repository.
philo-he 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 e875cda898 [GLUTEN-12909][VL] Add support for HiveGenericUDTF (#12910)
e875cda898 is described below
commit e875cda898a83c6b58682ba8de132b64d53d948b
Author: WangGuangxin <[email protected]>
AuthorDate: Wed Sep 16 11:36:13 2026 +0800
[GLUTEN-12909][VL] Add support for HiveGenericUDTF (#12910)
---
.../apache/spark/sql/hive/HiveUDFTransformer.scala | 4 +-
.../execution/GlutenHiveUDFTransformerSuite.scala | 136 +++++++++++++++++++++
2 files changed, 139 insertions(+), 1 deletion(-)
diff --git
a/gluten-substrait/src/main/scala/org/apache/spark/sql/hive/HiveUDFTransformer.scala
b/gluten-substrait/src/main/scala/org/apache/spark/sql/hive/HiveUDFTransformer.scala
index 4caf4fb43e..d30579fe2d 100644
---
a/gluten-substrait/src/main/scala/org/apache/spark/sql/hive/HiveUDFTransformer.scala
+++
b/gluten-substrait/src/main/scala/org/apache/spark/sql/hive/HiveUDFTransformer.scala
@@ -26,7 +26,7 @@ import java.util.Locale
object HiveUDFTransformer {
def isHiveUDF(expr: Expression): Boolean = {
expr match {
- case _: HiveSimpleUDF | _: HiveGenericUDF => true
+ case _: HiveSimpleUDF | _: HiveGenericUDF | _: HiveGenericUDTF => true
case _ => false
}
}
@@ -36,6 +36,8 @@ object HiveUDFTransformer {
(s.name.stripPrefix("default."), s.funcWrapper.functionClassName)
case g: HiveGenericUDF =>
(g.name.stripPrefix("default."), g.funcWrapper.functionClassName)
+ case t: HiveGenericUDTF =>
+ (t.name.stripPrefix("default."), t.funcWrapper.functionClassName)
case _ =>
throw new GlutenNotSupportException(
s"Expression $expr is not a HiveSimpleUDF or HiveGenericUDF")
diff --git
a/gluten-ut/test/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveUDFTransformerSuite.scala
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveUDFTransformerSuite.scala
new file mode 100644
index 0000000000..cbe019cbab
--- /dev/null
+++
b/gluten-ut/test/src/test/scala/org/apache/spark/sql/hive/execution/GlutenHiveUDFTransformerSuite.scala
@@ -0,0 +1,136 @@
+/*
+ * 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.spark.sql.hive.execution
+
+import org.apache.gluten.expression.{GenericExpressionTransformer, UDFMappings}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.internal.config.UI.UI_ENABLED
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.Generate
+import org.apache.spark.sql.hive.{HiveGenericUDTF, HiveUDFTransformer}
+import org.apache.spark.tags.SlowHiveTest
+
+import org.apache.commons.io.FileUtils
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF
+import org.apache.hadoop.hive.serde2.objectinspector.{ObjectInspector,
ObjectInspectorFactory, StructObjectInspector}
+import
org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory
+
+import java.io.File
+import java.nio.file.Files
+import java.util.Collections
+
+@SlowHiveTest
+class GlutenHiveUDFTransformerSuite extends SparkFunSuite {
+
+ private var baseDir: File = _
+ private var spark: SparkSession = _
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ if (baseDir == null) {
+ baseDir = Files.createTempDirectory(getClass.getSimpleName).toFile
+ }
+ if (spark == null) {
+ spark =
SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate()
+ spark.sparkContext.setLogLevel("warn")
+ }
+ }
+
+ override def afterAll(): Unit = {
+ try {
+ if (spark != null) {
+ spark.stop()
+ spark = null
+ }
+ if (baseDir != null) {
+ FileUtils.deleteDirectory(baseDir)
+ baseDir = null
+ }
+ } finally {
+ SparkSession.clearActiveSession()
+ SparkSession.clearDefaultSession()
+ super.afterAll()
+ }
+ }
+
+ private def sparkConf: SparkConf = {
+ val warehouseDir = new File(baseDir, "spark-warehouse").getAbsolutePath
+ val metastorePath = new File(baseDir, "metastore_db").getAbsolutePath
+
+ new SparkConf()
+ .set("spark.master", "local[1]")
+ .set("spark.app.name", getClass.getSimpleName)
+ .set("spark.ui.enabled", UI_ENABLED.defaultValueString)
+ .set("spark.sql.warehouse.dir", warehouseDir)
+ .set("javax.jdo.option.ConnectionURL",
s"jdbc:derby:;databaseName=$metastorePath;create=true")
+ }
+
+ test("HiveGenericUDTF is recognized and mapped by HiveUDFTransformer") {
+ spark.sql(
+ "CREATE TEMPORARY FUNCTION udtf_count2 " +
+ "AS 'org.apache.spark.sql.hive.execution.GlutenTestGenericUDTF'")
+
+ try {
+ val analyzed = spark.sql("SELECT udtf_count2(a) FROM (SELECT 1 AS a) t")
+ .queryExecution
+ .analyzed
+ val generate = analyzed
+ .collectFirst {
+ case generate: Generate if
generate.generator.isInstanceOf[HiveGenericUDTF] =>
+ generate
+ }
+ .getOrElse(fail(s"Expected HiveGenericUDTF in analyzed
plan:\n$analyzed"))
+ val udtf = generate.generator.asInstanceOf[HiveGenericUDTF]
+
+ assert(HiveUDFTransformer.isHiveUDF(udtf))
+ assert(
+ HiveUDFTransformer.getHiveUDFNameAndClassName(udtf) ===
+ ("udtf_count2",
"org.apache.spark.sql.hive.execution.GlutenTestGenericUDTF"))
+
+ val previousMapping = UDFMappings.hiveUDFMap.get("udtf_count2")
+ UDFMappings.hiveUDFMap.put("udtf_count2", "test_hive_generic_udtf")
+ try {
+ val transformer =
+ HiveUDFTransformer.replaceWithExpressionTransformer(udtf,
generate.child.output)
+ assert(transformer.isInstanceOf[GenericExpressionTransformer])
+ assert(transformer.substraitExprName === "test_hive_generic_udtf")
+ assert(transformer.original eq udtf)
+ } finally {
+ previousMapping match {
+ case Some(mapping) => UDFMappings.hiveUDFMap.put("udtf_count2",
mapping)
+ case None => UDFMappings.hiveUDFMap.remove("udtf_count2")
+ }
+ }
+ } finally {
+ spark.sql("DROP TEMPORARY FUNCTION IF EXISTS udtf_count2")
+ }
+ }
+}
+
+class GlutenTestGenericUDTF extends GenericUDTF {
+ override def initialize(arguments: Array[ObjectInspector]):
StructObjectInspector = {
+ ObjectInspectorFactory.getStandardStructObjectInspector(
+ Collections.singletonList("value"),
+
Collections.singletonList(PrimitiveObjectInspectorFactory.javaIntObjectInspector))
+ }
+
+ override def process(arguments: Array[AnyRef]): Unit = {}
+
+ override def close(): Unit = {}
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
