This is an automated email from the ASF dual-hosted git repository.
marin-ma 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 ccfacb07d0 [GLUTEN-13014][VL] Make a udfLibraryPaths UDF callable by
its own name (#13016)
ccfacb07d0 is described below
commit ccfacb07d09e3fea0b2ed0badde7fca96ab72a15
Author: Pedrum Jalali <[email protected]>
AuthorDate: Thu Sep 17 01:05:57 2026 -0700
[GLUTEN-13014][VL] Make a udfLibraryPaths UDF callable by its own name
(#13016)
---
.../gluten/backendsapi/velox/VeloxRuleApi.scala | 7 +-
.../org/apache/gluten/config/VeloxConfig.scala | 22 +++++++
.../apache/spark/sql/expression/UDFResolver.scala | 50 +++++++++++++-
.../gluten/expression/UDFResolverSuite.scala | 77 ++++++++++++++++++++++
.../apache/gluten/expression/VeloxUdfSuite.scala | 28 ++++++++
cpp/velox/udf/examples/MyUDF.cc | 38 +++++++++++
docs/developers/VeloxUDF.md | 15 +++++
docs/velox-configuration.md | 1 +
8 files changed, 236 insertions(+), 2 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
index 11d1944590..997d66b47e 100644
---
a/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
+++
b/backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala
@@ -17,7 +17,7 @@
package org.apache.gluten.backendsapi.velox
import org.apache.gluten.backendsapi.{BackendsApiManager, RuleApi}
-import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.config.{GlutenConfig, VeloxConfig}
import org.apache.gluten.extension._
import org.apache.gluten.extension.columnar._
import
org.apache.gluten.extension.columnar.MiscColumnarRules.{PreventBatchTypeMismatchInTableCache,
RemoveGlutenTableCacheColumnarToRow, RemoveTopmostColumnarToRow,
RewriteSubqueryBroadcast}
@@ -34,6 +34,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
import org.apache.spark.sql.execution._
import org.apache.spark.sql.execution.datasources.noop.GlutenNoopWriterRule
+import org.apache.spark.sql.expression.UDFResolver
class VeloxRuleApi extends RuleApi {
import VeloxRuleApi._
@@ -67,6 +68,10 @@ object VeloxRuleApi {
if (BackendsApiManager.getSettings.supportAppendDataExec()) {
injector.injectPlannerStrategy(SparkShimLoader.getSparkShims.getRewriteCreateTableAsSelect(_))
}
+
+ if (VeloxConfig.nativeUDFBypassRegistration) {
+ UDFResolver.getFunctionDescriptions.foreach(injector.injectFunction)
+ }
}
/**
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index 7fb3f69dd3..5d377a9368 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -16,7 +16,9 @@
*/
package org.apache.gluten.config
+import org.apache.spark.SparkEnv
import org.apache.spark.network.util.ByteUnit
+import org.apache.spark.sql.internal.SparkConfigUtil._
import org.apache.spark.sql.internal.SQLConf
import java.util.Locale
@@ -132,6 +134,16 @@ object VeloxConfig extends ConfigRegistry {
new VeloxConfig(SQLConf.get)
}
+ /**
+ * Reads the flag straight off the SparkConf instead of going through
[[get]].
+ *
+ * Session extensions are applied while the SparkSession is still being
built, so `SQLConf.get`
+ * returns defaults at that point and [[get]] would report this flag as off
however the user set
+ * it.
+ */
+ def nativeUDFBypassRegistration: Boolean =
+ Option(SparkEnv.get).exists(_.conf.get(NATIVE_UDF_BYPASS_REGISTRATION))
+
// velox caching options.
val COLUMNAR_VELOX_CACHE_ENABLED =
buildStaticConf("spark.gluten.sql.columnar.backend.velox.cacheEnabled")
@@ -696,6 +708,16 @@ object VeloxConfig extends ConfigRegistry {
.booleanConf
.createWithDefault(true)
+ val NATIVE_UDF_BYPASS_REGISTRATION =
+
buildStaticConf("spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration")
+ .doc(
+ "If true, a UDF from udfLibraryPaths can be called by the name it was
registered " +
+ "with, so you do not have to write a Java class for it or run CREATE
TEMPORARY " +
+ "FUNCTION. In exchange, there is no Java version to fall back to, so
any query " +
+ "Gluten cannot run natively will fail instead of running on Spark.
Off by default.")
+ .booleanConf
+ .createWithDefault(false)
+
val CAST_FROM_VARCHAR_ADD_TRIM_NODE =
buildConf("spark.gluten.velox.castFromVarcharAddTrimNode")
.doc(
diff --git
a/backends-velox/src/main/scala/org/apache/spark/sql/expression/UDFResolver.scala
b/backends-velox/src/main/scala/org/apache/spark/sql/expression/UDFResolver.scala
index 43ee7b4f7d..bad8303f21 100644
---
a/backends-velox/src/main/scala/org/apache/spark/sql/expression/UDFResolver.scala
+++
b/backends-velox/src/main/scala/org/apache/spark/sql/expression/UDFResolver.scala
@@ -19,13 +19,16 @@ package org.apache.spark.sql.expression
import org.apache.gluten.backendsapi.velox.VeloxBackendSettings
import org.apache.gluten.exception.{GlutenException, GlutenNotSupportException}
import org.apache.gluten.expression._
+import org.apache.gluten.extension.injector.FunctionDescription
import org.apache.gluten.jni.JniWorkspace
import org.apache.spark.{SparkConf, SparkFiles}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
+import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast,
Expression, Unevaluable}
+import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast,
Expression, ExpressionInfo, Unevaluable}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction
import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
@@ -37,6 +40,7 @@ import org.apache.spark.util.Utils
import java.io.File
import java.net.URI
import java.nio.file.{Files, FileVisitOption, Paths}
+import java.util.Locale
import scala.collection.JavaConverters.asScalaIteratorConverter
import scala.collection.mutable
@@ -338,6 +342,50 @@ object UDFResolver extends Logging {
.toBoolean
}
+ /**
+ * One Spark function per loaded UDF whose name contains no dot. A dotted
name is a Hive UDF class
+ * name, which VeloxHiveUDFTransformer already resolves, so it is skipped
here.
+ *
+ * A name is also skipped when it collides with a Spark built-in: the names
are unqualified, so
+ * injecting one would redirect that built-in to a native implementation
with possibly different
+ * semantics for every query on the session.
+ *
+ * Names differing only in case are skipped as a group. Spark lowercases a
function name when it
+ * registers it, so they would collapse to one entry and the last
registration would win, leaving
+ * a call to either name running the other one's implementation.
+ */
+ def getFunctionDescriptions: Seq[FunctionDescription] = {
+ val candidates = UDFNames.toSeq.filterNot(_.contains(".")).sorted
+ val byLowerCase = candidates.groupBy(_.toLowerCase(Locale.ROOT))
+
+ val (ambiguous, distinct) =
+ candidates.partition(name =>
byLowerCase(name.toLowerCase(Locale.ROOT)).size > 1)
+
+ val (shadowing, injectable) =
+ distinct.partition(name =>
FunctionRegistry.builtin.functionExists(FunctionIdentifier(name)))
+
+ ambiguous.foreach(
+ name =>
+ logWarning(
+ s"Not registering UDF '$name' by name: it differs only in case from
another UDF " +
+ s"loaded from the same libraries, and Spark function names are
case-insensitive. " +
+ s"Rename it in the UDF library to call it directly."))
+
+ shadowing.foreach(
+ name =>
+ logWarning(
+ s"Not registering UDF '$name' by name: it shadows a Spark built-in.
" +
+ s"Rename it in the UDF library to call it directly."))
+
+ injectable.map {
+ name =>
+ (
+ FunctionIdentifier(name),
+ new ExpressionInfo(classOf[UDFExpression].getName, name),
+ (children: Seq[Expression]) => getUdfExpression(name,
name)(children))
+ }
+ }
+
def getUdfExpression(name: String, alias: String)(children:
Seq[Expression]): UDFExpression = {
def errorMessage: String =
s"UDF $name -> ${children.map(_.dataType.simpleString).mkString(", ")}
is not registered."
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/expression/UDFResolverSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/expression/UDFResolverSuite.scala
new file mode 100644
index 0000000000..6968be7da2
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/gluten/expression/UDFResolverSuite.scala
@@ -0,0 +1,77 @@
+/*
+ * 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.expression
+
+import org.apache.gluten.config.VeloxConfig
+
+import org.apache.spark.sql.expression.UDFResolver
+
+import org.scalatest.BeforeAndAfterEach
+import org.scalatest.funsuite.AnyFunSuite
+
+class UDFResolverSuite extends AnyFunSuite with BeforeAndAfterEach {
+
+ // UDFNames is JVM-global and populated once per JVM, so it is restored
rather than cleared.
+ private var savedNames: Set[String] = Set.empty
+
+ override protected def beforeEach(): Unit = {
+ savedNames = UDFResolver.UDFNames.toSet
+ UDFResolver.UDFNames.clear()
+ }
+
+ override protected def afterEach(): Unit = {
+ UDFResolver.UDFNames.clear()
+ UDFResolver.UDFNames ++= savedNames
+ }
+
+ private def describedNames(): Seq[String] =
+ UDFResolver.getFunctionDescriptions.map(_._1.funcName)
+
+ test("registration by name is off unless it is turned on") {
+
assert(VeloxConfig.NATIVE_UDF_BYPASS_REGISTRATION.defaultValue.contains(false))
+ }
+
+ test("a name with no dot is described") {
+ UDFResolver.UDFNames += "myudf_increment"
+ assert(describedNames() == Seq("myudf_increment"))
+ }
+
+ test("a dotted name is skipped, it is a hive udf class name") {
+ UDFResolver.UDFNames +=
"org.apache.spark.sql.hive.execution.UDFStringString"
+ assert(describedNames().isEmpty)
+ }
+
+ test("a name colliding with a spark built-in is skipped") {
+ UDFResolver.UDFNames += "abs"
+ assert(describedNames().isEmpty)
+ }
+
+ test("a name colliding with a spark built-in does not skip the others") {
+ UDFResolver.UDFNames ++= Seq("abs", "myudf_increment", "upper")
+ assert(describedNames() == Seq("myudf_increment"))
+ }
+
+ test("names differing only in case are skipped as a group") {
+ UDFResolver.UDFNames ++= Seq("Foo", "foo", "myudf_increment")
+ assert(describedNames() == Seq("myudf_increment"))
+ }
+
+ test("names are described in sorted order") {
+ UDFResolver.UDFNames ++= Seq("b_udf", "a_udf")
+ assert(describedNames() == Seq("a_udf", "b_udf"))
+ }
+}
diff --git
a/backends-velox/src/test/scala/org/apache/gluten/expression/VeloxUdfSuite.scala
b/backends-velox/src/test/scala/org/apache/gluten/expression/VeloxUdfSuite.scala
index a5128f62d9..493e1c79a7 100644
---
a/backends-velox/src/test/scala/org/apache/gluten/expression/VeloxUdfSuite.scala
+++
b/backends-velox/src/test/scala/org/apache/gluten/expression/VeloxUdfSuite.scala
@@ -17,12 +17,14 @@
package org.apache.gluten.expression
import org.apache.gluten.backendsapi.velox.VeloxBackendSettings
+import org.apache.gluten.config.VeloxConfig
import org.apache.gluten.execution.ProjectExecTransformer
import org.apache.gluten.execution.WindowExecTransformer
import org.apache.gluten.tags.{SkipTest, UDFTest}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{GlutenQueryTest, Row, SparkSession}
+import org.apache.spark.sql.catalyst.FunctionIdentifier
import org.apache.spark.sql.catalyst.plans.SQLHelper
import org.apache.spark.sql.execution.ProjectExec
import org.apache.spark.sql.execution.window.WindowExec
@@ -294,6 +296,30 @@ abstract class VeloxUdfSuite extends GlutenQueryTest with
SQLHelper {
}
}
}
+
+ test("native udf with a plain name is callable without a hive udf class") {
+ // No CREATE TEMPORARY FUNCTION and no Java class: the session extension
put the name in
+ // Spark's registry via SparkInjector.injectFunction, which is the only
writer for it.
+ assert(
+ spark.sessionState.functionRegistry
+ .lookupFunction(FunctionIdentifier("myudf_plus_one"))
+ .isDefined)
+
+ val df = spark.sql("SELECT myudf_plus_one(col1) FROM VALUES (1L), (2L),
(3L) AS t(col1)")
+ checkGlutenPlan[ProjectExecTransformer](df)
+ checkAnswer(df, Seq(Row(2L), Row(3L), Row(4L)))
+ }
+
+ test("native udf with a plain name fails at analysis when gluten is
disabled") {
+ withSQLConf(("spark.gluten.enabled", "false")) {
+ val e = intercept[Exception] {
+ spark.sql("SELECT myudf_plus_one(col1) FROM VALUES (1L) AS
t(col1)").collect()
+ }
+ // The injected function has no JVM implementation to fall back to, so
the call is
+ // rejected rather than silently returning a result from somewhere else.
+ assert(e.getMessage.contains("myudf_plus_one"))
+ }
+ }
}
@UDFTest
@@ -305,6 +331,8 @@ class VeloxUdfSuiteLocal extends VeloxUdfSuite {
.set("spark.files", udfLibPath)
.set(VeloxBackendSettings.GLUTEN_VELOX_UDF_LIB_PATHS, udfLibRelativePath)
.set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ // Off by default, so the by-name tests below have to opt in.
+ .set(VeloxConfig.NATIVE_UDF_BYPASS_REGISTRATION.key, "true")
}
}
diff --git a/cpp/velox/udf/examples/MyUDF.cc b/cpp/velox/udf/examples/MyUDF.cc
index 260629fdf9..687cdd2d3b 100644
--- a/cpp/velox/udf/examples/MyUDF.cc
+++ b/cpp/velox/udf/examples/MyUDF.cc
@@ -71,6 +71,43 @@ class HiveStringStringRegisterer final : public
gluten::UdfRegisterer {
} // namespace hivestringstring
+namespace myudfplusone {
+
+template <typename T>
+struct MyUdfPlusOneFunction {
+ VELOX_DEFINE_FUNCTION_TYPES(T);
+
+ FOLLY_ALWAYS_INLINE void call(int64_t& result, const int64_t& a) {
+ result = a + 1;
+ }
+};
+
+// name: myudf_plus_one
+// signatures:
+// bigint -> bigint
+// type: SimpleFunction
+// A name with no dot, so it is callable directly without a Hive UDF class.
+class MyUdfPlusOneRegisterer final : public gluten::UdfRegisterer {
+ public:
+ int getNumUdf() override {
+ return 1;
+ }
+
+ void populateUdfEntries(int& index, gluten::UdfEntry* udfEntries) override {
+ udfEntries[index++] = {name_.c_str(), kBigInt, 1, arg_, false, false};
+ }
+
+ void registerSignatures() override {
+ facebook::velox::registerFunction<MyUdfPlusOneFunction, int64_t,
int64_t>({name_});
+ }
+
+ private:
+ const std::string name_ = "myudf_plus_one";
+ const char* arg_[1] = {kBigInt};
+};
+
+} // namespace myudfplusone
+
std::vector<std::shared_ptr<gluten::UdfRegisterer>>& globalRegisters() {
static std::vector<std::shared_ptr<gluten::UdfRegisterer>> registerers;
return registerers;
@@ -83,6 +120,7 @@ void setupRegisterers() {
}
auto& registerers = globalRegisters();
registerers.push_back(std::make_shared<hivestringstring::HiveStringStringRegisterer>());
+
registerers.push_back(std::make_shared<myudfplusone::MyUdfPlusOneRegisterer>());
inited = true;
}
} // namespace
diff --git a/docs/developers/VeloxUDF.md b/docs/developers/VeloxUDF.md
index a38f1a48db..8dbf276a48 100644
--- a/docs/developers/VeloxUDF.md
+++ b/docs/developers/VeloxUDF.md
@@ -192,6 +192,20 @@ VeloxColumnarToRow
+- Scan hive spark_catalog.default.tbl [col1#11], HiveTableRelation
[`spark_catalog`.`default`.`tbl`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, Data Cols: [col1#11],
Partition Cols: []]
```
+## Natively Only UDF Registration
+
+This is an alternative to the registration described above, for a UDF that is
implemented only in Velox and has no Java counterpart.
+
+This is off by default. Set
`spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration=true` to
turn it on.
+
+Once enabled, a UDF whose registered name contains no dot is added to the
session's function registry under that name, provided the name is not already a
Spark built-in and no other loaded UDF differs from it only in case. It needs
no matching Hive UDF class, no jar on the classpath, and no `CREATE TEMPORARY
FUNCTION` — register it under a name with no dot, such as `my_udf`, and call it
directly:
+
+```
+spark-sql (default)> select my_udf(col1) from tbl;
+```
+
+**There is no fallback with this method.** A name registered this way has no
Java implementation behind it, so a query that Gluten cannot offload fails
instead of falling back to the JVM. Use the registration described above
whenever the fallback path is required.
+
## Configurations
| Parameters | Description
|
@@ -199,6 +213,7 @@ VeloxColumnarToRow
| spark.gluten.sql.columnar.backend.velox.udfLibraryPaths | Path to the
udf/udaf libraries.
|
| spark.gluten.sql.columnar.backend.velox.driver.udfLibraryPaths | Path to the
udf/udaf libraries on driver node. Only applicable on yarn-client mode.
|
| spark.gluten.sql.columnar.backend.velox.udfAllowTypeConversion | Whether to
inject possible `cast` to convert mismatched data types from input to one
registered signatures. |
+| spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration | Call
a UDF by the name it was registered with, without writing a Java class for it.
There is then no Java version to fall back to, so a query Gluten cannot run
natively will fail. Defaults to `false`. |
# Pandas UDFs (a.k.a. Vectorized UDFs)
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 2c14b37c45..3ab3220ad8 100644
--- a/docs/velox-configuration.md
+++ b/docs/velox-configuration.md
@@ -61,6 +61,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.memInitCapacity
| 🔄 Dynamic | 8MB | The initial memory capacity to reserve
for a newly created Velox query memory pool.
[...]
|
spark.gluten.sql.columnar.backend.velox.memoryPoolCapacityTransferAcrossTasks
| 🔄 Dynamic | true | Whether to allow memory capacity transfer
between memory pools from different tasks.
[...]
| spark.gluten.sql.columnar.backend.velox.memoryUseHugePages
| 🔄 Dynamic | false | Use explicit huge pages for Velox
memory allocation.
[...]
+| spark.gluten.sql.columnar.backend.velox.nativeUDF.bypassRegistration
| ⚓ Static | false | If true, a UDF from udfLibraryPaths
can be called by the name it was registered with, so you do not have to write a
Java class for it or run CREATE TEMPORARY FUNCTION. In exchange, there is no
Java version to fall back to, so any query Gluten cannot run natively will fail
instead of running on Spark. Off by default.
[...]
| spark.gluten.sql.columnar.backend.velox.numCacheFileHandles
| ⚓ Static | 10000 | Maximum number of entries in the file
handle cache. Each entry holds an open file descriptor (local FS) or connection
state (remote FS). Note that on local filesystems, high values may approach the
OS file descriptor limit (ulimit -n). On remote object stores (S3, ABFS, GCS)
entries represent network connections/sockets rather than per-file OS file
descriptors, but the [...]
| spark.gluten.sql.columnar.backend.velox.orc.scan.enabled
| 🔄 Dynamic | true | Enable velox orc scan. If disabled,
vanilla spark orc scan will be used.
[...]
| spark.gluten.sql.columnar.backend.velox.parquet.dictionaryPageSizeBytes
| 🔄 Dynamic | 2MB | The maximum size in bytes for a
Parquet dictionary page
[...]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
