voonhous commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3968383260
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -410,9 +504,19 @@ object HoodieProcedureFilterUtils {
case Success(result) => result
// 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.
- case Failure(e @ (_: ArithmeticException | _: NumberFormatException | _:
DateTimeException)) => throw e
+ // string; each extends the matching JDK type. 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 of which
the hardcoded table
+ // never reached before the registry fallback existed. Swallowing any of
these would
+ // silently drop a row the same query keeps, so let them out and let the
caller fail the
+ // way the equivalent query does. Only for an otherwise-evaluable
expression, though: a
+ // caller that skips validateFilterExpression and evaluates a genuinely
unsupported function
+ // directly still no-matches (calling eval() on the leftover
UnresolvedFunction/Unevaluable
+ // node raises the same SparkThrowable-family INTERNAL_ERROR for an
entirely different,
+ // expected reason), so this method stays safe to call on its own.
+ case Failure(e @ (_: ArithmeticException | _: NumberFormatException | _:
DateTimeException
+ | _: SparkThrowable | _: IllegalArgumentException)) if
!boundExpr.exists(_.isInstanceOf[Unevaluable]) =>
Review Comment:
Addressed at 467472bdeed7: `ArithmeticException` / `NumberFormatException` /
`DateTimeException` are back on an unconditional arm, and the
`!boundExpr.exists(_.isInstanceOf[Unevaluable])` guard now covers only the two
newly reachable types.
Verified on 3.5.5 with ANSI on: evaluating `int(name) > 1 OR
no_such_fn(name) = 'x'` directly returned 0 rows at 97c30d35dfd7 and raises
`SparkNumberFormatException` at this head. The new case at
`TestHoodieProcedureFilterUtils:203` pins it.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +416,73 @@ object HoodieProcedureFilterUtils {
}
}
+ // Resolves a function not covered by the hardcoded table above via Spark's
own FunctionRegistry.
+ // A resolved result is only usable if it can actually be eval()'d one row
at a time, which
+ // several categories of otherwise-valid expressions cannot:
RuntimeReplaceable placeholders
+ // (nvl, ifnull, left, right) need substitution the analyzer normally
performs but skips here,
+ // and can themselves unwrap to another RuntimeReplaceable (regexp_substr ->
NullIf) so the
+ // unwrap has to run to a fixed point; aggregates (percentile, collect_list)
only make sense
+ // across real aggregation; generators (explode, inline) only work inside a
projection;
+ // non-deterministic functions (rand, uuid, spark_partition_id) expect
per-partition
+ // initialization; window/grouping-only builtins (current_user, lag, lead,
...) are Unevaluable
+ // outside their normal context; and a type mismatch the analyzer's
implicit-cast pass would
+ // normally have caught still fails checkInputDataTypes - checked both on
the raw lookup result
+ // (its own declared input-type contract, e.g. split_part's, is otherwise
discarded once
+ // unwrapped) and again after unwrapping and widening (e.g. nvl(ts, 0) only
becomes checkable
+ // once it's the Coalesce(ts, 0) the hardcoded coalesce(ts, 0) case would
already have widened).
+ // Anything in one of those categories is treated as still-unresolved so it
falls through to the
+ // existing rejection path instead of silently dropping every row.
+ private def resolveViaFunctionRegistry(unresolvedFunc: UnresolvedFunction,
sparkSession: SparkSession): Expression = {
+ Try {
+ // 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.
+ val resolved = unresolvedFunc.nameParts match {
+ case Seq(funcName) =>
+
sparkSession.sessionState.functionRegistry.lookupFunction(FunctionIdentifier(funcName),
unresolvedFunc.arguments)
+ case _ => unresolvedFunc
+ }
+ // lookupFunction alone skips the analyzer's own implicit-cast rule, so
a wrapper declaring
+ // a real input-type contract (nvl needing matching operand types,
split_part needing
+ // string/string/int, ...) sees its raw, uncast arguments here. Casting
via that same rule
+ // before checking the contract lets a fixable mismatch (nvl(ts, 0), a
Long/Int pair) widen
+ // the way coalesce(ts, 0) already does, while a genuine mismatch
(split_part's delimiter
+ // passed as Int, which nothing implicit-casts to String) still fails as
it should.
+ val castedResolved = applyImplicitCasts(resolved)
+ // Checked against checkInputDataTypes only, not the resolved flag: a
RuntimeReplaceable
Review Comment:
Addressed at 467472bdeed7 - the block is down to why the unwrap runs to a
fixed point and why the result still needs a usability verdict, with the rest
moved onto `lookupBuiltin` (:447), `finalizeRegistryResolution` (:460) and
`isUsableOutsideQueryPlan` (:480), next to the checks they explain.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +416,73 @@ object HoodieProcedureFilterUtils {
}
}
+ // Resolves a function not covered by the hardcoded table above via Spark's
own FunctionRegistry.
+ // A resolved result is only usable if it can actually be eval()'d one row
at a time, which
+ // several categories of otherwise-valid expressions cannot:
RuntimeReplaceable placeholders
+ // (nvl, ifnull, left, right) need substitution the analyzer normally
performs but skips here,
+ // and can themselves unwrap to another RuntimeReplaceable (regexp_substr ->
NullIf) so the
+ // unwrap has to run to a fixed point; aggregates (percentile, collect_list)
only make sense
+ // across real aggregation; generators (explode, inline) only work inside a
projection;
+ // non-deterministic functions (rand, uuid, spark_partition_id) expect
per-partition
+ // initialization; window/grouping-only builtins (current_user, lag, lead,
...) are Unevaluable
+ // outside their normal context; and a type mismatch the analyzer's
implicit-cast pass would
+ // normally have caught still fails checkInputDataTypes - checked both on
the raw lookup result
+ // (its own declared input-type contract, e.g. split_part's, is otherwise
discarded once
+ // unwrapped) and again after unwrapping and widening (e.g. nvl(ts, 0) only
becomes checkable
+ // once it's the Coalesce(ts, 0) the hardcoded coalesce(ts, 0) case would
already have widened).
+ // Anything in one of those categories is treated as still-unresolved so it
falls through to the
+ // existing rejection path instead of silently dropping every row.
+ private def resolveViaFunctionRegistry(unresolvedFunc: UnresolvedFunction,
sparkSession: SparkSession): Expression = {
+ Try {
+ // 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.
+ val resolved = unresolvedFunc.nameParts match {
+ case Seq(funcName) =>
+
sparkSession.sessionState.functionRegistry.lookupFunction(FunctionIdentifier(funcName),
unresolvedFunc.arguments)
+ case _ => unresolvedFunc
+ }
Review Comment:
Addressed at 467472bdeed7: `lookupBuiltin` (:447),
`finalizeRegistryResolution` (:460, unwrap-to-fixed-point plus coercion) and
`isUsableOutsideQueryPlan` (:480) are separate helpers now, so
`resolveViaFunctionRegistry` reads as lookup -> implicit casts -> contract
check -> finalize -> usability check.
##########
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:
Addressed at 467472bdeed7, and the cross-version answer turns out to be
narrower than the whole 4.x line: `CurrentTimestampLike implements
FoldableUnevaluable` only on 4.0.x. javap over the catalyst jars gives
`CodegenFallback` on 3.5.5, 4.1.1 and 4.2.0, and `FoldableUnevaluable` on 4.0.1
and 4.0.2.
Probed with the head's own predicate against `FunctionRegistry.builtin`: on
4.0.2, `current_timestamp` / `now` / `localtimestamp` / `current_date` are
`foldable=true`, not `Unevaluable`, and `eval(EmptyRow)` throws
`[INTERNAL_ERROR] Cannot evaluate expression` - so the new foldable probe
rejects them there instead of letting the rethrow at :521 fail the procedure
call. On 4.1.1 / 4.2.0 they eval fine and stay supported. The crash is closed.
Residual: 4.0 is now the odd one out. `current_timestamp() > t` evaluates on
3.5 / 4.1 / 4.2 and is rejected on 4.0, while
`TestHoodieProcedureFilterUtils:436` pins the evaluating case unconditionally,
so that assertion would fail on a spark4.0 build. The spark4.0 lanes are
commented out in `bot.yml` ([CI-TRIM]), so CI will not show it. Could we gate
that line on `HoodieSparkUtils` the way :216 already does, with `gteqSpark4_0
&& !gteqSpark4_1` as the rejecting window?
--
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]
