hudi-agent commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3965671676
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -411,8 +500,17 @@ object HoodieProcedureFilterUtils {
// Spark raises SparkArithmeticException for an overflowing ANSI cast or
arithmetic, and
// SparkNumberFormatException or SparkDateTimeException for an ANSI cast
of a malformed
// string; each extends the matching JDK type. Swallowing one would
silently drop a row the
- // same query keeps, so let it out and let the caller fail the way the
equivalent query does.
+ // same query keeps, so let it out unconditionally, exactly as before
the registry fallback.
case Failure(e @ (_: ArithmeticException | _: NumberFormatException | _:
DateTimeException)) => throw e
+ // SparkThrowable covers the equivalent runtime errors from
registry-resolved functions
+ // (to_number/bit_get out-of-range, ...), and IllegalArgumentException
covers a bad regex
+ // pattern - both newly reachable through the registry fallback, so this
guard only applies
+ // to these two: a caller that skips validateFilterExpression and
evaluates a genuinely
+ // unsupported function directly still no-matches instead of hitting the
same
+ // SparkThrowable-family INTERNAL_ERROR that Unevaluable.eval() raises
for an unrelated
+ // reason, so this method stays safe to call on its own.
+ case Failure(e @ (_: SparkThrowable | _: IllegalArgumentException)) if
!boundExpr.exists(_.isInstanceOf[Unevaluable]) =>
Review Comment:
🤖 The comment says these are "newly reachable through the registry
fallback", but `PatternSyntaxException` is an `IllegalArgumentException`, so
this also changes hardcoded-table functions: `rlike(name, '[')` or
`regexp_extract` with a bad pattern used to drop rows and now aborts the whole
call mid-iteration. That looks like the right parity with Spark to me — just
want to confirm it's an intended widening rather than a side effect.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +416,68 @@ object HoodieProcedureFilterUtils {
}
}
+ // Resolves a function not covered by the hardcoded table above via Spark's
own FunctionRegistry,
+ // then checks the result is actually usable outside a real query plan -
both steps a plain
+ // lookupFunction call skips or can't tell on its own. Anything that isn't
falls through to the
+ // existing rejection path (see #19850) instead of letting eval() throw
silently.
+ private def resolveViaFunctionRegistry(unresolvedFunc: UnresolvedFunction,
sparkSession: SparkSession): Expression = {
+ Try {
+ val castedResolved = applyImplicitCasts(lookupBuiltin(unresolvedFunc,
sparkSession))
+ // Checked here, on the raw wrapper, before unwrapping: a
RuntimeReplaceable wrapper's own
+ // declared input-type contract (nvl needing matching operand types,
split_part needing
+ // string/string/int) is otherwise discarded once unwrapped to a form
with a weaker or
+ // absent contract of its own.
+ if (!castedResolved.checkInputDataTypes().isSuccess) {
+ unresolvedFunc
+ } else {
+ val finalized = finalizeRegistryResolution(castedResolved)
+ if (isUsableOutsideQueryPlan(finalized)) finalized else unresolvedFunc
+ }
+ } match {
+ case Success(resolved) => resolved
+ case Failure(_) => unresolvedFunc
+ }
+ }
+
+ // Filter expressions only ever call plain builtins. FunctionRegistry
registers builtins with no
+ // database, so a db-qualified or 3+ part name (db.func, catalog.db.func)
can only be resolved
+ // by guessing which part is the real function name - that risks matching an
unrelated
+ // same-named function, so those are left unresolved instead.
+ private def lookupBuiltin(unresolvedFunc: UnresolvedFunction, sparkSession:
SparkSession): Expression =
+ unresolvedFunc.nameParts match {
+ case Seq(funcName) =>
+
sparkSession.sessionState.functionRegistry.lookupFunction(FunctionIdentifier(funcName),
unresolvedFunc.arguments)
+ case _ => unresolvedFunc
+ }
+
+ // RuntimeReplaceable placeholders (nvl, ifnull, left, right, ...) need
substitution the analyzer
+ // normally performs but lookupFunction skips, and can themselves unwrap to
another
+ // RuntimeReplaceable (regexp_substr -> NullIf), so the unwrap runs to a
fixed point. Then widens
+ // numeric operands the same way pass three would - nvl(ts, 0) unwraps to
Coalesce(ts, 0), which
+ // needs the same widening the hardcoded coalesce(ts, 0) case gets - so a
registry function and
+ // its hardcoded-table equivalent agree on what counts as resolved.
+ private def finalizeRegistryResolution(expression: Expression): Expression =
{
+ def unwrapReplacements(expr: Expression): Expression = {
+ val next = expr.transformUp { case r: RuntimeReplaceable =>
r.replacement }
+ if (next.fastEquals(expr)) next else unwrapReplacements(next)
+ }
+ applyCoercionRules(unwrapReplacements(expression))
+ }
+
+ // A resolved expression still isn't usable one row at a time if it's an
aggregate (percentile,
+ // collect_list - only make sense across real aggregation), a generator
(explode, inline - only
+ // work inside a projection), still Unevaluable somewhere in it
(current_user, lag, lead, ... -
+ // only valid in their normal analyzer context), or non-deterministic (rand,
uuid,
+ // spark_partition_id - expect per-partition initialization this evaluator
never does).
+ private def isUsableOutsideQueryPlan(expression: Expression): Boolean = {
+
!expression.isInstanceOf[org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction]
&&
+
!expression.isInstanceOf[org.apache.spark.sql.catalyst.expressions.Generator] &&
+ !expression.exists(_.isInstanceOf[Unevaluable]) &&
Review Comment:
🤖 `Unevaluable` isn't the whole "can't eval outside a plan" family on Spark
4: there `Unevaluable extends FoldableUnevaluable`, and `CurrentTimestampLike`
(current_timestamp/now/localtimestamp) mixes in `FoldableUnevaluable` only — so
it passes this guard, validates OK, then `eval()` throws INTERNAL_ERROR, which
the new rethrow at line 512 propagates and fails the whole procedure call.
Could you check this against the spark4.0 profile? Matching on the trait name
(or explicitly rejecting the current_timestamp family) would keep 3.x and 4.x
in agreement. @yihua for the cross-version call.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -92,10 +99,13 @@ object HoodieProcedureFilterUtils {
}
}
- // Second pass: resolve functions
- val functionResolved = attributeBound.transform {
+ // Second pass: resolve functions. transformUp so a nested call's
arguments (e.g. upper(name)
+ // inside instr(upper(name), 'A')) are already resolved by the time the
outer function's case
+ // runs - otherwise resolved/checkInputDataTypes below would see an
unresolved child and
+ // reject a call that's actually fine.
+ val functionResolved = attributeBound.transformUp {
case unresolvedFunc:
org.apache.spark.sql.catalyst.analysis.UnresolvedFunction =>
- unresolvedFunc.nameParts.head.toLowerCase(Locale.ROOT) match {
+ val hardcodedResolved =
unresolvedFunc.nameParts.head.toLowerCase(Locale.ROOT) match {
Review Comment:
🤖 nit: the case body now computes `hardcodedResolved` via a large match and
then immediately matches on that result again to decide whether to fall back to
the registry — could this second match be pulled into a small named helper
(e.g. `resolveOrFallback(hardcodedResolved, unresolvedFunc, sparkSession)`) so
the transformUp case reads as a single call instead of a match nested in a
match?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
