This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new cf2f9914cd [spark] Support REST catalog Hive UDTFs in lateral views
(#10510)
cf2f9914cd is described below
commit cf2f9914cdebf5c0062c200f82e7a3b43b9540d8
Author: Zouxxyy <[email protected]>
AuthorDate: Wed Oct 14 10:02:17 2026 +0800
[spark] Support REST catalog Hive UDTFs in lateral views (#10510)
---
docs/docs/spark/sql-functions.md | 9 +-
.../catalyst/analysis/PaimonFunctionResolver.scala | 40 +++--
.../extensions/RewritePaimonFunctionCommands.scala | 115 ++++++++++---
.../spark/sql/PaimonV1FunctionTestBase.scala | 188 ++++++++++++++++++++-
4 files changed, 310 insertions(+), 42 deletions(-)
diff --git a/docs/docs/spark/sql-functions.md b/docs/docs/spark/sql-functions.md
index f2d638d7da..357ac219ce 100644
--- a/docs/docs/spark/sql-functions.md
+++ b/docs/docs/spark/sql-functions.md
@@ -145,7 +145,7 @@ Paimon-managed function definitions require a REST catalog.
Choose a definition
| Definition | Use it for | Requirement |
| --- | --- | --- |
| [Java lambda](#lambda-function) | A small Java expression stored in the
catalog | Register metadata and a Spark definition with procedures. |
-| [JAR implementation](#file-function) | Existing Spark or Hive UDF/UDAF
classes | Spark 3.4+ and the implementation JAR. |
+| [JAR implementation](#file-function) | Spark or Hive UDF/UDAF classes, or
Hive UDTF classes | Spark 3.4+ and the implementation JAR. |
| [SQL body](#sql-function) | Reusable scalar SQL logic | Spark 4.0+. |
The examples assume the REST catalog and its `default` database are selected.
@@ -177,9 +177,12 @@ CALL sys.drop_function(`function` => 'default.area_func');
### File Function
-Register a Spark or Hive UDF/UDAF implementation packaged in a JAR.
+Register a Spark or Hive UDF/UDAF, or a Hive UDTF, packaged in a JAR.
+See [Spark
UDFs](https://spark.apache.org/docs/latest/sql-ref-functions.html#udfs-user-defined-functions).
-Currently, supports Spark or Hive implementations of UDFs and UDAFs, see
[Spark
UDFs](https://spark.apache.org/docs/latest/sql-ref-functions.html#udfs-user-defined-functions)
+Hive UDTFs can be invoked in a `SELECT` list or a `LATERAL VIEW`. In `LATERAL
VIEW`,
+use an unqualified function name or `database.function` in the current REST
catalog;
+Spark's parser does not accept `catalog.database.function` in this clause.
This feature requires Spark 3.4 or higher.
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFunctionResolver.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFunctionResolver.scala
index 3d4ee94876..fb9161a777 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFunctionResolver.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFunctionResolver.scala
@@ -25,10 +25,10 @@ import
org.apache.paimon.spark.catalog.functions.PaimonFunctions
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.UnresolvedFunction
import org.apache.spark.sql.catalyst.expressions.{Cast, Expression, Literal}
-import
org.apache.spark.sql.catalyst.parser.extensions.UnResolvedPaimonV1Function
+import
org.apache.spark.sql.catalyst.parser.extensions.{UnResolvedPaimonV1Function,
UnResolvedPaimonV1Generator}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_FUNCTION
+import org.apache.spark.sql.catalyst.trees.TreePattern.{GENERATOR,
UNRESOLVED_FUNCTION}
import org.apache.spark.sql.connector.catalog.CatalogPlugin
import org.apache.spark.sql.types.{BinaryType, DataType, DayTimeIntervalType,
LongType, NullType, StringType}
import org.apache.spark.unsafe.types.UTF8String
@@ -40,9 +40,9 @@ case class PaimonFunctionResolver(spark: SparkSession)
extends Rule[LogicalPlan]
protected lazy val catalogManager = spark.sessionState.catalogManager
override def apply(plan: LogicalPlan): LogicalPlan =
-
plan.resolveOperatorsUpWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION)) {
+
plan.resolveOperatorsUpWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION,
GENERATOR)) {
case l: LogicalPlan =>
-
l.transformExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION)) {
+
l.transformExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION,
GENERATOR)) {
case u: UnresolvedFunction
if isDescriptorToPresignedUrlFunction(u.nameParts) &&
u.arguments.forall(_.resolved) =>
@@ -51,20 +51,30 @@ case class PaimonFunctionResolver(spark: SparkSession)
extends Rule[LogicalPlan]
if isBlobViewFunction(u.nameParts) &&
u.arguments.forall(_.resolved) =>
resolveBlobView(u)
case u: UnResolvedPaimonV1Function if u.arguments.forall(_.resolved)
=>
- u.funcIdent.catalog match {
- case Some(catalog) =>
- catalogManager.catalog(catalog) match {
- case v1FunctionCatalog: SupportV1Function =>
- v1FunctionCatalog.registerAndResolveV1Function(u)
- case _ =>
- throw new IllegalArgumentException(
- s"Catalog $catalog is not a v1 function catalog")
- }
- case None => throw new IllegalArgumentException("Catalog name is
not defined")
- }
+ resolveV1Function(u)
+ case u: UnResolvedPaimonV1Generator if u.childrenResolved =>
+ val function = UnResolvedPaimonV1Function(
+ u.funcIdent,
+ u.arguments,
+ isDistinct = false,
+ func = u.func)
+ u.resolve(resolveV1Function(function))
}
}
+ private def resolveV1Function(u: UnResolvedPaimonV1Function): Expression = {
+ u.funcIdent.catalog match {
+ case Some(catalog) =>
+ catalogManager.catalog(catalog) match {
+ case v1FunctionCatalog: SupportV1Function =>
+ v1FunctionCatalog.registerAndResolveV1Function(u)
+ case _ =>
+ throw new IllegalArgumentException(s"Catalog $catalog is not a v1
function catalog")
+ }
+ case None => throw new IllegalArgumentException("Catalog name is not
defined")
+ }
+ }
+
private def resolveBlobView(u: UnresolvedFunction): Expression = {
if (u.arguments.length != 3) {
throw new UnsupportedOperationException(
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
index 6be680cd60..43f9a49d92 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewritePaimonFunctionCommands.scala
@@ -25,16 +25,18 @@ import
org.apache.paimon.spark.execution.{CreatePaimonV1FunctionCommand, Describ
import org.apache.paimon.spark.util.OptionUtils
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.FunctionIdentifier
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedException,
UnresolvedFunction}
+import org.apache.spark.sql.catalyst.{FunctionIdentifier, InternalRow}
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedException,
UnresolvedFunction, UnresolvedGenerator}
import org.apache.spark.sql.catalyst.catalog.CatalogFunction
-import org.apache.spark.sql.catalyst.expressions.{Expression, Unevaluable}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Generator,
SubqueryExpression, Unevaluable}
+import org.apache.spark.sql.catalyst.expressions.codegen.{CodegenContext,
ExprCode}
import org.apache.spark.sql.catalyst.plans.logical.{CreateFunction,
DescribeFunction, DropFunction, LogicalPlan, SubqueryAlias, UnresolvedWith}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.catalyst.trees.TreePattern.{TreePattern,
UNRESOLVED_FUNCTION}
+import org.apache.spark.sql.catalyst.trees.TreePattern.{GENERATOR,
PLAN_EXPRESSION, TreePattern, UNRESOLVED_FUNCTION}
import org.apache.spark.sql.connector.catalog.CatalogManager
+import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
import org.apache.spark.sql.paimon.shims.SparkShimLoader
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, StructType}
case class RewritePaimonFunctionCommands(spark: SparkSession) extends
Rule[LogicalPlan] {
@@ -123,28 +125,95 @@ case class RewritePaimonFunctionCommands(spark:
SparkSession) extends Rule[Logic
u,
alias =>
transformPaimonV1Function(alias).asInstanceOf[SubqueryAlias])
case l: LogicalPlan =>
-
l.transformExpressionsWithPruning(_.containsAnyPattern(UNRESOLVED_FUNCTION)) {
+ l.transformExpressionsWithPruning(
+ _.containsAnyPattern(UNRESOLVED_FUNCTION, GENERATOR,
PLAN_EXPRESSION)) {
+ case subquery: SubqueryExpression =>
+ // Spark's recursive subquery analyzer does not rerun this parser
rewrite.
+ subquery.withNewPlan(transformPaimonV1Function(subquery.plan))
case u: UnresolvedFunction =>
- lookup.CatalogAndFunctionIdentifier.unapply(u.nameParts) match {
- case Some((v1FunctionCatalog: SupportV1Function, funcIdent,
false))
- // For Paimon built-in functions, Spark will resolve them by
itself.
- if !lookup.isPaimonBuiltInFunction(funcIdent) =>
- // If the function is already registered, avoid redundant
lookup in the catalog to reduce overhead.
- if (v1FunctionCatalog.v1FunctionRegistered(funcIdent)) {
- UnResolvedPaimonV1Function(funcIdent, u, None)
- } else {
- val function = v1FunctionCatalog.getFunction(funcIdent)
- if (PaimonFunctionLookup.isPaimonV1Function(function)) {
- UnResolvedPaimonV1Function(funcIdent, u, Some(function))
- } else {
- u
- }
- }
- case _ => u
- }
+ lookupFunction(u.nameParts)
+ .map {
+ case (funcIdent, function) =>
UnResolvedPaimonV1Function(funcIdent, u, function)
+ }
+ .getOrElse(u)
+ case u @ UnresolvedGenerator(name, arguments) =>
+ // Spark parses a two-part generator name as database.function.
Qualify it before
+ // catalog lookup so a database with the same name as a catalog is
not reinterpreted.
+ val nameParts = name.database
+ .map(
+ database =>
+ Seq(
+
name.catalog.getOrElse(catalogManager.currentCatalog.name()),
+ database,
+ name.funcName))
+ .getOrElse(name.nameParts)
+ lookupFunction(nameParts)
+ .map {
+ case (funcIdent, function) =>
+ UnResolvedPaimonV1Generator(funcIdent, arguments, function)
+ }
+ .getOrElse(u)
}
}
}
+
+ private def lookupFunction(
+ nameParts: Seq[String]): Option[(FunctionIdentifier,
Option[PaimonFunction])] = {
+ lookup.CatalogAndFunctionIdentifier.unapply(nameParts) match {
+ case Some((v1FunctionCatalog: SupportV1Function, funcIdent, false))
+ // Spark resolves built-in functions itself.
+ if !lookup.isPaimonBuiltInFunction(funcIdent) =>
+ // Avoid redundant catalog lookups for functions already loaded into
the registry.
+ if (v1FunctionCatalog.v1FunctionRegistered(funcIdent)) {
+ Some((funcIdent, None))
+ } else {
+ val function = v1FunctionCatalog.getFunction(funcIdent)
+ if (PaimonFunctionLookup.isPaimonV1Function(function)) {
+ Some((funcIdent, Some(function)))
+ } else {
+ None
+ }
+ }
+ case _ => None
+ }
+ }
+}
+
+/** Keeps a REST catalog generator unresolved until Spark resolves its
arguments. */
+case class UnResolvedPaimonV1Generator(
+ funcIdent: FunctionIdentifier,
+ arguments: Seq[Expression],
+ func: Option[PaimonFunction])
+ extends Generator {
+
+ override def children: Seq[Expression] = arguments
+
+ override def elementSchema: StructType = throw new
UnresolvedException("elementSchema")
+ override def dataType: DataType = throw new UnresolvedException("dataType")
+ override def foldable: Boolean = throw new UnresolvedException("foldable")
+ override def nullable: Boolean = throw new UnresolvedException("nullable")
+ override lazy val resolved = false
+
+ override def prettyName: String = funcIdent.unquotedString
+
+ def resolve(function: Expression): Generator = function match {
+ case generator: Generator => generator
+ case other =>
+ throw QueryCompilationErrors.generatorNotExpectedError(
+ funcIdent,
+ other.getClass.getCanonicalName)
+ }
+
+ // Nothing is compatible with both Scala 2.12's TraversableOnce and 2.13's
IterableOnce.
+ override def eval(input: InternalRow): Nothing =
+ throw QueryExecutionErrors.cannotEvaluateExpressionError(this)
+
+ override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode =
+ throw QueryExecutionErrors.cannotGenerateCodeForExpressionError(this)
+
+ override protected def withNewChildrenInternal(
+ newChildren: IndexedSeq[Expression]): UnResolvedPaimonV1Generator =
+ copy(arguments = newChildren)
}
/** An unresolved Paimon V1 function to let Spark resolve the necessary
variables. */
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonV1FunctionTestBase.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonV1FunctionTestBase.scala
index 367327ed47..c0144b9e3f 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonV1FunctionTestBase.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PaimonV1FunctionTestBase.scala
@@ -22,12 +22,198 @@ import
org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
import org.apache.paimon.spark.function.FunctionResources._
import org.apache.spark.SparkConf
-import org.apache.spark.sql.Row
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.expressions.Explode
import java.io.File
abstract class PaimonV1FunctionTestBase extends
PaimonSparkTestWithRestCatalogBase {
+ private val hivePosExplode =
"org.apache.hadoop.hive.ql.udf.generic.GenericUDTFPosExplode"
+
+ test("Paimon V1 Function: persistent Hive UDTF in lateral view") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ withTable("udtf_input") {
+ sql("CREATE TABLE udtf_input (id INT, items ARRAY<INT>) USING paimon")
+ sql("INSERT INTO udtf_input VALUES (1, array(10, 20)), (2, array()),
(3, NULL)")
+
+ // Resolve a cold REST catalog function before any SELECT invocation
populates its registry.
+ Seq("udtf_pos_explode", "test.udtf_pos_explode").foreach {
+ name =>
+ checkAnswer(
+ sql(s"""SELECT id, position, item FROM udtf_input
+ |LATERAL VIEW $name(items) u AS position, item
+ |""".stripMargin),
+ Seq(Row(1, 0, 10), Row(1, 1, 20))
+ )
+ }
+
+ checkAnswer(
+ sql("""SELECT id, position, item FROM udtf_input
+ |LATERAL VIEW OUTER udtf_pos_explode(items) u AS position, item
+ |""".stripMargin),
+ Seq(Row(1, 0, 10), Row(1, 1, 20), Row(2, null, null), Row(3, null,
null))
+ )
+ checkAnswer(
+ sql("SELECT id, paimon.test.udtf_pos_explode(items) FROM
udtf_input"),
+ Seq(Row(1, 0, 10), Row(1, 1, 20)))
+
+ val session = spark.newSession()
+ session.sql("USE paimon.test")
+ checkAnswer(
+ session.sql("""SELECT id, position, item FROM udtf_input
+ |LATERAL VIEW udtf_pos_explode(items) u AS position,
item
+ |""".stripMargin),
+ Seq(Row(1, 0, 10), Row(1, 1, 20))
+ )
+ }
+ }
+ }
+
+ test("Paimon V1 Function: lateral view database name can match a catalog") {
+ withDatabase("paimon.paimon") {
+ sql("CREATE DATABASE paimon.paimon")
+ withUserDefinedFunction("paimon.paimon.udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION paimon.paimon.udtf_pos_explode AS
'$hivePosExplode'")
+ checkAnswer(
+ sql("""SELECT position, item FROM VALUES (array(10, 20)) AS t(items)
+ |LATERAL VIEW paimon.udtf_pos_explode(items) u AS position,
item
+ |""".stripMargin),
+ Seq(Row(0, 10), Row(1, 20))
+ )
+ }
+ }
+ }
+
+ test("Paimon V1 Function: Hive UDTF in CTE and dependent lateral views") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ checkAnswer(
+ sql("""WITH expanded AS (
+ | SELECT id, item FROM VALUES (1, array(10, 20)) AS t(id, items)
+ | LATERAL VIEW udtf_pos_explode(items) u AS position, item
+ |)
+ |SELECT * FROM expanded
+ |""".stripMargin),
+ Seq(Row(1, 10), Row(1, 20))
+ )
+ checkAnswer(
+ sql("""SELECT id, a.item, b.item FROM VALUES (1, array(10, 20)) AS
t(id, items)
+ |LATERAL VIEW udtf_pos_explode(items) a AS position, item
+ |LATERAL VIEW udtf_pos_explode(array(a.item, a.item + 1)) b AS
position, item
+ |""".stripMargin),
+ Seq(Row(1, 10, 10), Row(1, 10, 11), Row(1, 20, 20), Row(1, 20, 21))
+ )
+ }
+ }
+
+ test("Paimon V1 Function: Hive UDTF in scalar subquery") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ Seq("posexplode", "udtf_pos_explode").foreach {
+ name =>
+ checkAnswer(
+ sql(s"""SELECT (
+ | SELECT count(*) FROM VALUES (array(10, 20)) AS t(items)
+ | LATERAL VIEW $name(items) u AS position, item
+ |) AS n
+ |""".stripMargin),
+ Seq(Row(2L))
+ )
+ }
+ }
+ }
+
+ test("Paimon V1 Function: Hive UDTF in EXISTS subquery") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ Seq(10 -> true, 30 -> false).foreach {
+ case (item, expected) =>
+ checkAnswer(
+ sql(s"""SELECT EXISTS (
+ | SELECT 1 FROM VALUES (array(10, 20)) AS t(items)
+ | LATERAL VIEW test.udtf_pos_explode(items) u AS position,
item
+ | WHERE u.item = $item
+ |)
+ |""".stripMargin),
+ Seq(Row(expected))
+ )
+ }
+ checkAnswer(
+ sql("""SELECT id FROM VALUES (10), (30) AS outer_t(id)
+ |WHERE EXISTS (
+ | SELECT 1 FROM VALUES (array(10, 20)) AS t(items)
+ | LATERAL VIEW udtf_pos_explode(items) u AS position, item
+ | WHERE u.item = outer_t.id
+ |)
+ |""".stripMargin),
+ Seq(Row(10))
+ )
+ }
+ }
+
+ test("Paimon V1 Function: Hive UDTF in nested expression subqueries and
CTEs") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ checkAnswer(
+ sql("""SELECT (
+ | WITH expanded AS (
+ | SELECT (
+ | SELECT count(*) FROM VALUES (array(10, 20)) AS t(items)
+ | LATERAL VIEW udtf_pos_explode(items) u AS position, item
+ | ) AS n
+ | )
+ | SELECT n FROM expanded
+ |) AS n
+ |""".stripMargin),
+ Seq(Row(2L))
+ )
+ }
+ }
+
+ test("Paimon V1 Function: lateral view preserves built-in and temporary
function lookup") {
+ withUserDefinedFunction("udtf_pos_explode" -> false) {
+ sql(s"CREATE FUNCTION udtf_pos_explode AS '$hivePosExplode'")
+ withUserDefinedFunction("udtf_pos_explode" -> true) {
+ spark.sessionState.functionRegistry.createOrReplaceTempFunction(
+ "udtf_pos_explode",
+ children => Explode(children.head),
+ "scala_udf")
+ checkAnswer(
+ sql("""SELECT item FROM VALUES (array(10, 20)) AS t(items)
+ |LATERAL VIEW udtf_pos_explode(items) u AS item
+ |""".stripMargin),
+ Seq(Row(10), Row(20))
+ )
+ checkAnswer(
+ sql("""SELECT position, item FROM VALUES (array(10, 20)) AS t(items)
+ |LATERAL VIEW test.udtf_pos_explode(items) u AS position, item
+ |""".stripMargin),
+ Seq(Row(0, 10), Row(1, 20))
+ )
+ checkAnswer(
+ sql("""SELECT position, item FROM VALUES (array(10, 20)) AS t(items)
+ |LATERAL VIEW posexplode(items) u AS position, item
+ |""".stripMargin),
+ Seq(Row(0, 10), Row(1, 20))
+ )
+ }
+ }
+ }
+
+ test("Paimon V1 Function: reject a scalar function in lateral view") {
+ withUserDefinedFunction("udf_add2" -> false) {
+ sql(s"CREATE FUNCTION udf_add2 AS '$UDFExampleAdd2Class' USING JAR
'$testUDFJarPath'")
+ val error = intercept[AnalysisException] {
+ sql("""SELECT item FROM VALUES (1, 2) AS t(a, b)
+ |LATERAL VIEW udf_add2(a, b) u AS item
+ |""".stripMargin)
+ }
+ assert(error.getErrorClass == "UNSUPPORTED_GENERATOR.NOT_GENERATOR")
+ }
+ }
+
test("Paimon V1 Function: create or replace function") {
withUserDefinedFunction("udf_add2" -> false) {
sql(s"""
