hudi-agent commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3963069880
##########
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)
Review Comment:
🤖 `applyImplicitCasts` only runs `ImplicitTypeCasts`, but the analyzer runs
the whole `TypeCoercion` rule set — so calls that depend on any of the other
rules get rejected here even though Spark accepts them. `concat(id, 'x')` is
the clearest one: `ConcatCoercion` casts non-binary children to string, so
`select concat(1, 'x')` returns `1x` in Spark 3.x, but here
`Concat.checkInputDataTypes` fails and the filter is rejected. `if(cond, ts,
0)` (`IfCoercion`) and `greatest`/`least`/`array_contains`
(`FunctionArgumentConversion`) look like they'd hit the same thing. Could the
pass run the fuller rule list, or at least the handful most likely to show up
in filters?
<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:
##########
@@ -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:
🤖 The new `if !boundExpr.exists(_.isInstanceOf[Unevaluable])` guard also
applies to the three pre-existing ANSI cases, which were rethrown
unconditionally before. So `cast(name as int) = 1 OR no_such_fn(name) = 'x'`
under ANSI now swallows the `NumberFormatException` and drops the row instead
of surfacing it. Reachable only when a caller skips `validateFilterExpression`,
but was the narrowing of the existing three types intentional, or should the
guard apply only to the two newly-added types?
<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,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:
🤖 nit: this comment block (and the one on `resolveViaFunctionRegistry`
itself) is quite dense to parse in one pass — might be worth trimming to the
couple of non-obvious points (why RuntimeReplaceable needs a fixed-point
unwrap, why 3+ part names are rejected) and moving the rest into shorter inline
comments near the specific checks they explain.
<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,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:
🤖 nit: `resolveViaFunctionRegistry` now does lookup, implicit-cast, a first
checkInputDataTypes, RuntimeReplaceable unwrap-to-fixed-point, coercion, and a
second suitability check all in one method — might be worth pulling the
unwrap-to-fixed-point + "still unsupported" check into its own small helper
(e.g. `finalizeRegistryResolution`) so each step reads as a single
responsibility.
<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]
