hudi-agent commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3970231585
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +417,76 @@ 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))
Review Comment:
🤖 This first `checkInputDataTypes` runs before the argument subtree has been
widened, so a registry function over a mixed-type arithmetic argument gets
rejected: `sqrt(ts + 1) > 0` or `concat(name, ts + 1) = 'a11001'` (ts is Long,
1 is Int) leaves `Add` unresolved, `ImplicitTypeCasts`/`ConcatCoercion` skip on
`!childrenResolved`, and the Long child fails the input-type check ->
"Unsupported functions: sqrt", while the same shape through the hardcoded table
(`abs(ts + 1)`) works because pass three widens afterwards. Would running
`applyCoercionRules` over `unresolvedFunc.arguments` before the lookup close
that gap?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +371,92 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Seq(rows(1)))(keep(rows, "`50% overlap` > 15", schema))
}
- test("evaluateFilter silently drops rows for expressions it cannot resolve")
{
- assertResult(Seq.empty)(keep(scalarRows, "concat(name, 'x') = 'a1x'",
scalarSchema))
- assertResult(Seq.empty)(keep(scalarRows, "instr(name, 'a') = 1",
scalarSchema))
- assertResult(Seq.empty)(keep(scalarRows, "if(name = 'a1', true, false)",
scalarSchema))
+ test("evaluateFilter resolves functions outside the hardcoded table via
FunctionRegistry") {
+ // Functions missing from the hardcoded table now fall back to Spark's own
FunctionRegistry
+ // instead of being rejected as unsupported. See #19852.
+ assertKeeps(scalarRows, "concat(name, 'x') = 'a1x'", Seq(scalarRows.head))
+ assertKeeps(scalarRows, "instr(name, 'a') = 1", Seq(scalarRows.head))
+ assertKeeps(scalarRows, "if(name = 'a1', true, false)",
Seq(scalarRows.head))
assertResult(Seq(scalarRows.head))(
keep(scalarRows, "case when name = 'a1' then true else false end",
scalarSchema))
// Or short-circuits on the resolved side, which is what the
unresolved-operand guard preserves.
- assertResult(Seq(scalarRows.head))(
- keep(scalarRows, "id = 1 OR concat(name, 'x') = 'a1x'", scalarSchema))
+ assertKeeps(scalarRows, "id = 1 OR concat(name, 'x') = 'a1x'",
Seq(scalarRows.head))
+
+ // RuntimeReplaceable builtins (nvl, left, right, ...) resolve to a
placeholder node that
+ // FunctionRegistry.lookupFunction doesn't substitute on its own - make
sure we unwrap it
+ // rather than letting eval() blow up on the raw placeholder.
+ assertKeeps(scalarRows, "nvl(name, 'z') = 'a1'", Seq(scalarRows.head))
+ assertKeeps(scalarRows, "left(name, 1) = 'a'", Seq(scalarRows.head))
+ assertKeeps(scalarRows, "right(name, 1) = '1'", Seq(scalarRows.head))
+
+ // A hardcoded-table entry called with an arity the table doesn't handle
(substring only
+ // handles 3 args) should still fall back to the registry instead of
getting stuck.
+ assertKeeps(scalarRows, "substring(name, 2) = '1'", Seq(scalarRows.head))
+
+ // The rejection message for multiple unknown functions lists every name,
sorted.
+ assert(validate("no_such_fn(name) = 'x' OR other_missing(name) = 1")
+ .left.exists(_ == "Unsupported functions: no_such_fn, other_missing"))
+
+ // A 3+ part name (catalog.db.func) isn't safe to look up by bare function
name alone - make
+ // sure it's rejected rather than silently resolved against a same-named
function elsewhere.
+ assert(validate("some_catalog.some_db.upper(name) = 'A1'").isLeft)
+ assertResult(Seq.empty)(keep(scalarRows, "some_catalog.some_db.upper(name)
= 'A1'", scalarSchema))
+ // Same story for a 2-part db-qualified name: builtins register with no
database, so
+ // FunctionRegistry has no "default.upper" to find, and guessing by
dropping the qualifier
+ // would risk the same wrong-function-match problem as the 3+ part case.
+ assert(validate("default.upper(name) = 'A1'").isLeft)
+ assertResult(Seq.empty)(keep(scalarRows, "default.upper(name) = 'A1'",
scalarSchema))
+ }
+
+ test("evaluateFilter still rejects aggregate/generator/nondeterministic
functions resolved via FunctionRegistry") {
+ // Aggregate functions resolve fine as expressions but can't be eval()'d
per row - make sure
+ // those still go through the existing #19850 rejection path instead of
silently resolving to
+ // a broken, always-false filter. Same story for generators (explode only
makes sense in a
+ // projection) and non-deterministic functions (rand()/uuid() rely on
per-partition
+ // initialization this evaluator never does). any_value is covered
separately below - the
+ // parser lowers it straight to an AggregateExpression before it ever
reaches this guard.
+ // max(id) is an unambiguous AggregateFunction case (no decimal-literal
argument to complicate
+ // why it's rejected, unlike percentile's 0.5), so it's what actually pins
the guard clause.
+ assert(validate("max(id) > 0").left.exists(_.contains("Unsupported
functions: max")))
+ assertResult(Seq.empty)(keep(scalarRows, "max(id) > 0", scalarSchema))
+ assert(validate("percentile(id, 0.5) = 1").isLeft)
+ assert(validate("explode(array(1, 2)) =
1").left.exists(_.contains("Unsupported functions: explode")))
+ assert(validate("rand() = 1").isLeft)
+ assert(validate("uuid() = 'x'").isLeft)
+ assertResult(Seq.empty)(keep(scalarRows, "rand() = 1", scalarSchema))
+ // monotonically_increasing_id/input_file_name are also Nondeterministic,
so the same
+ // deterministic check catches them without needing their own case.
+ assert(validate("monotonically_increasing_id() = 1").isLeft)
+ assert(validate("input_file_name() = 'x'").isLeft)
+ // current_timestamp is deterministic-at-eval-time (Spark computes it
directly rather than
+ // requiring rule substitution), so it resolves and evaluates for real
instead of needing
+ // denylist treatment. current_date doesn't share that: it's a
TimeZoneAwareExpression that
+ // stays unresolved without a session zone the same way hour(t) does
above, not because of
+ // anything this guard rejects.
+ assertResult(scalarRows)(keep(scalarRows, "current_timestamp() > t",
scalarSchema))
Review Comment:
🤖 On Spark 4.0.2 `CurrentTimestampLike` implements `FoldableUnevaluable`
(and `Unevaluable extends FoldableUnevaluable`, not the reverse), so its `eval`
throws, the foldable probe in `isUsableOutsideQueryPlan` rejects it, and `keep`
returns `Seq.empty` here rather than `scalarRows`. The 4.0 CI lane is commented
out right now so this won't surface, but under `-Pspark4.0` this assertion
fails - could it be gated like the `gteqSpark4_0` check at line 216, or assert
the rejection on 4.0?
<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 +417,76 @@ 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
Review Comment:
🤖 nit: `resolveOrFallback` takes both the already-computed
`hardcodedResolved` and the original `unresolvedFunc` just to re-check the same
match — might be worth renaming the params (e.g. `firstAttempt`/`original`) so
it's clearer at the call site which is which.
<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 +417,76 @@ 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
+ }
Review Comment:
🤖 nit: `resolveViaFunctionRegistry` chains lookup, implicit-casts,
input-type check, unwrap/coercion, and a second usability check all in one Try
block — have you considered splitting the "resolve" step from the "is this
usable standalone" check so each concern is easier to follow/test on its own?
<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]
