voonhous commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3961138518
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -348,7 +349,7 @@ object HoodieProcedureFilterUtils {
} else {
unresolvedFunc
}
- case _ => unresolvedFunc
+ case _ => resolveViaFunctionRegistry(unresolvedFunc, sparkSession)
Review Comment:
Addressed: the registry fallback now runs on whatever the hardcoded match
returns if it is still an `UnresolvedFunction`, so a wrong-arity call on a
table entry falls through as well. `substring(name, 2) = '1'` is pinned as
resolving.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -367,12 +376,41 @@ object HoodieProcedureFilterUtils {
}
}
- private def evaluateExpressionOnRow(expression: Expression, row: Row,
schema: StructType): Boolean = {
+ // didn't match anything above, so ask Spark itself before we give up -
saves us from having
+ // to hand-list every builtin (concat, instr, if, ...) one by one
+ private def resolveViaFunctionRegistry(unresolvedFunc: UnresolvedFunction,
sparkSession: SparkSession): Expression = {
+ Try {
+ val nameParts = unresolvedFunc.nameParts
+ val functionIdentifier = nameParts match {
+ case Seq(funcName) => FunctionIdentifier(funcName)
+ case Seq(db, funcName) => FunctionIdentifier(funcName, Some(db))
+ case _ => FunctionIdentifier(nameParts.last)
+ }
+ val resolved =
sparkSession.sessionState.functionRegistry.lookupFunction(functionIdentifier,
unresolvedFunc.arguments)
+ // lookupFunction alone doesn't run the analyzer rule that swaps these
placeholders for
+ // their real expression (nvl, ifnull, left, right, ...) - eval() on the
raw node just
+ // throws, so unwrap it ourselves.
+ val unwrapped = resolved.transformUp { case r: RuntimeReplaceable =>
r.replacement }
+ // aggregate functions (percentile, collect_list, ...) resolve fine here
but can't be
+ // eval()'d row-by-row outside of real aggregation - treat them as
still-unresolved so
+ // the existing rejection path (see #19850) catches them instead of
silently no-matching.
+ if
(unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction])
{
Review Comment:
Addressed: the guard now also rejects `Generator` results and anything with
`!deterministic`. `explode`, `rand`, `uuid`, `monotonically_increasing_id` and
`input_file_name` are pinned as rejected; `current_timestamp()` still evaluates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
