voonhous commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3961552119
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
Review Comment:
**major:** Confirmed, and it is not only `nvl`. On the `show_cleans` output
schema (`time_taken_in_millis` is `LongType`):
| filter | Spark | this PR |
|---|---|---|
| `nvl(time_taken_in_millis, 0) > 100` | 1 row | rejected: `Unsupported
functions: nvl` |
| `nvl(total_files_deleted, 0) > 1` (Int column) | 1 row | 1 row |
| `nvl(time_taken_in_millis, 0L) > 100L` | 1 row | 1 row |
| `coalesce(time_taken_in_millis, 0) > 100` | 1 row | 1 row |
Same for `if(time_taken_in_millis > 100, ...)`, `greatest` and `nullif`: the
`!unwrapped.resolved` check at line 432 sees `Coalesce(Long, Int)` before the
third pass widens it. Not a regression (the base rejected every registry name),
so major rather than blocking. Could we run the third-pass coercion over
`resolved` before the `stillUnsupported` check, so registry functions widen the
same way the hardcoded ones do?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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;
+ // 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; and a type
mismatch the analyzer's
+ // implicit-cast pass would normally have caught (e.g. concat on a
non-string column) still
+ // fails checkInputDataTypes. 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 or db-qualified builtins. A
3+ part name
+ // (catalog.db.func) isn't safe to look up: FunctionIdentifier only
carries one qualifier,
+ // and guessing by dropping the extra parts risks matching an unrelated
same-named function.
+ val functionIdentifier = unresolvedFunc.nameParts match {
+ case Seq(funcName) => Some(FunctionIdentifier(funcName))
+ case Seq(db, funcName) => Some(FunctionIdentifier(funcName, Some(db)))
+ case _ => None
+ }
+ val resolved = functionIdentifier
+ .map(sparkSession.sessionState.functionRegistry.lookupFunction(_,
unresolvedFunc.arguments))
+ .getOrElse(unresolvedFunc)
+ val unwrapped = resolved.transformUp { case r: RuntimeReplaceable =>
r.replacement }
Review Comment:
**blocker:** A `RuntimeReplaceable` whose replacement is itself
`RuntimeReplaceable` survives this single `transformUp`, because the pass does
not revisit the node it just substituted. `regexp_substr(action, 'lean') =
'lean'` on the `show_cleans` output then passes validation and returns 0 rows
where Spark returns 2 (`RegExpSubStr` unwraps to a `NullIf` whose `eval`
throws, swallowed at line 469); the base rejected it with `Unsupported
functions: regexp_substr`. `RuntimeReplaceable` does not extend `Unevaluable`,
so no other guard catches it. Could we iterate the unwrap to a fixed point?
```suggestion
def unwrap(expr: Expression): Expression = {
val next = expr.transformUp { case r: RuntimeReplaceable =>
r.replacement }
if (next.fastEquals(expr)) next else unwrap(next)
}
val unwrapped = unwrap(resolved)
```
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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
Review Comment:
**major:** This holds for analysis-time failures only. A registry function
that throws at eval time falls through to `case Failure(_) => false` at line
469, so `to_number(action, '999') > 0`, `bit_get(version, 99) = 0` and
`regexp_replace(action, '[', 'x') = 'x'` now pass validation and return an
empty result where Spark raises; the base rejected all three names. The class
pre-exists for `regexp_extract`, but the reachable surface grows from about 34
names to 167. Could we widen the rethrow at line 469 to `SparkThrowable` and
`IllegalArgumentException` (or drop the blanket `false`), and pin one
invalid-regex filter?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +354,64 @@ 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(name, 'x') =
'a1x'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(name, 'a') =
1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "if(name = 'a1', true,
false)", scalarSchema))
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))
+ assertResult(Right(()))(validate("concat(name, 'x') = 'a1x'"))
+ assertResult(Right(()))(validate("instr(name, 'a') = 1"))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "nvl(name, 'z') =
'a1'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "left(name, 1) = 'a'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "right(name, 1) =
'1'", scalarSchema))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2) =
'1'", scalarSchema))
+
+ // 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))
+ }
+
+ test("evaluateFilter still rejects aggregate/generator/nondeterministic
functions resolved via FunctionRegistry") {
Review Comment:
**major:** Three of the five guard clauses are not pinned here: with the
`AggregateFunction`, `Generator` or `checkInputDataTypes` clause deleted, every
assertion in this test still passes. `percentile(id, 0.5)` is rejected only
because the decimal literal leaves it unresolved, `any_value` is lowered by the
parser and never enters the fallback, `explode` only changes message, and
`rand()`/`uuid()` are unresolved anyway (only
`monotonically_increasing_id`/`input_file_name` pin the deterministic clause).
Could we use `max(id) > 0` with `.left.exists(_.contains("Unsupported
functions: max"))` and assert the `explode` message the same way?
`checkInputDataTypes` is implied by `resolved` for every builtin, so that
clause can go.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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;
+ // 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; and a type
mismatch the analyzer's
+ // implicit-cast pass would normally have caught (e.g. concat on a
non-string column) still
+ // fails checkInputDataTypes. 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 or db-qualified builtins. A
3+ part name
+ // (catalog.db.func) isn't safe to look up: FunctionIdentifier only
carries one qualifier,
+ // and guessing by dropping the extra parts risks matching an unrelated
same-named function.
+ val functionIdentifier = unresolvedFunc.nameParts match {
+ case Seq(funcName) => Some(FunctionIdentifier(funcName))
+ case Seq(db, funcName) => Some(FunctionIdentifier(funcName, Some(db)))
+ case _ => None
+ }
+ val resolved = functionIdentifier
+ .map(sparkSession.sessionState.functionRegistry.lookupFunction(_,
unresolvedFunc.arguments))
+ .getOrElse(unresolvedFunc)
+ val unwrapped = resolved.transformUp { case r: RuntimeReplaceable =>
r.replacement }
+ val stillUnsupported =
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction]
||
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.Generator] ||
+ !unwrapped.deterministic ||
+ !unwrapped.resolved ||
+ !unwrapped.checkInputDataTypes().isSuccess
Review Comment:
**major:** The guards run on `unwrapped`, so the input-type contract
declared on the discarded wrapper is never checked. `SplitPart` declares
`ImplicitCastInputTypes` over `(String, String, Int)`, but its replacement
`ElementAt(StringSplitSQL(...))` declares none, so `split_part(action, 1, 1) =
'clean'` passes validation, throws `ClassCastException` per row (swallowed at
line 469) and returns 0 rows where Spark returns 2; the base rejected it. Could
we evaluate `resolved` and `checkInputDataTypes` on `resolved` (pre-unwrap),
keeping the unwrap only for the returned expression? That still accepts `left`,
`nvl`, `nullif` and `char_length`.
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -538,12 +587,10 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
}
test("validateFilterExpression rejects expressions the evaluator cannot
resolve") {
- val unknown = validate("concat(name, 'x') = 'a1x' OR instr(name, 'a') = 1")
- assert(unknown.left.exists(_.contains("Unsupported functions: concat,
instr")))
-
- assert(validate("if(name = 'a1', true, false)").isLeft)
- assert(validate("substring(name, 2)").isLeft)
- assert(validate("id = 1 OR concat(name, 'x') =
'a1x'").left.exists(_.contains("Unsupported functions: concat")))
+ // concat/instr/substring(2-arg) now resolve via the FunctionRegistry
fallback (see #19852),
+ // so they're no longer rejected here — covered by the "resolves functions
... via
Review Comment:
**minor:** Two things went with the removed assertions. The multi-name
message (`Unsupported functions: concat, instr`, sorted and comma-joined) now
has no coverage anywhere, and the positive registry cases above (`nvl`, `left`,
`right`, `substring(name, 2)`, `if`) are asserted through `keep` only, never
through `validate`, although every procedure validates before it evaluates. Not
blocking. Could we re-add the multi-name check with two unknown names
(`no_such_fn(name) = 'x' OR other_missing(name) = 1`) and pair each positive
`keep` with `assertResult(Right(()))(validate(...))`, perhaps via a small
`assertKeeps` helper?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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;
+ // 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; and a type
mismatch the analyzer's
+ // implicit-cast pass would normally have caught (e.g. concat on a
non-string column) still
+ // fails checkInputDataTypes. 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 or db-qualified builtins. A
3+ part name
+ // (catalog.db.func) isn't safe to look up: FunctionIdentifier only
carries one qualifier,
+ // and guessing by dropping the extra parts risks matching an unrelated
same-named function.
+ val functionIdentifier = unresolvedFunc.nameParts match {
+ case Seq(funcName) => Some(FunctionIdentifier(funcName))
+ case Seq(db, funcName) => Some(FunctionIdentifier(funcName, Some(db)))
Review Comment:
**minor:** This branch is dead: builtins are registered with `database =
None` (`FunctionRegistry.internalRegisterFunction`), so
`lookupFunction(FunctionIdentifier("upper", Some("default")), ...)` throws
`UNRESOLVED_ROUTINE` and `default.upper(name) = 'A1'` ends up rejected exactly
like the 3-part case, only without a test. Not blocking. Could we fold it into
the `None` arm (or route it through `sessionState.catalog.lookupFunction` if
db-qualified names are meant to work), and add the `default.upper(...)`
assertion next to the `some_catalog.some_db.upper(...)` one?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestShowCleansProcedures.scala:
##########
@@ -637,9 +637,11 @@ class TestShowCleansProcedures extends
HoodieSparkProcedureTestBase {
s"""call show_clean_plans(table => '$tableName', filter =>
"nonexistent_col > 1")""")(
"Invalid column references: nonexistent_col")
+ // concat is now resolved via the FunctionRegistry fallback (see
#19852), so a genuinely
+ // unknown function name is needed here to exercise the rejection path.
checkExceptionContain(
- s"""call show_clean_plans(table => '$tableName', filter =>
"concat(action, 'x') = 'cleanx'")""")(
- "Unsupported functions: concat")
+ s"""call show_clean_plans(table => '$tableName', filter =>
"no_such_fn(action) = 'cleanx'")""")(
Review Comment:
**minor:** This is the only procedure-level coverage of the change, and it
is the negative case. The `filterTests` list at lines 458-470 only calls
`UPPER` and `LENGTH`, both in the hardcoded table, so nothing at the `call`
level shows a registry-resolved function working end to end. Not blocking.
Could we add one row there, e.g. `("concat(action, 'x') = 'cleanx'",
"Registry-resolved function")`? Before this PR that threw at validation, so the
existing `length > 0` assertion at line 477 discriminates.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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;
+ // 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; and a type
mismatch the analyzer's
+ // implicit-cast pass would normally have caught (e.g. concat on a
non-string column) still
+ // fails checkInputDataTypes. 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 or db-qualified builtins. A
3+ part name
+ // (catalog.db.func) isn't safe to look up: FunctionIdentifier only
carries one qualifier,
+ // and guessing by dropping the extra parts risks matching an unrelated
same-named function.
+ val functionIdentifier = unresolvedFunc.nameParts match {
+ case Seq(funcName) => Some(FunctionIdentifier(funcName))
+ case Seq(db, funcName) => Some(FunctionIdentifier(funcName, Some(db)))
+ case _ => None
+ }
+ val resolved = functionIdentifier
+ .map(sparkSession.sessionState.functionRegistry.lookupFunction(_,
unresolvedFunc.arguments))
+ .getOrElse(unresolvedFunc)
+ val unwrapped = resolved.transformUp { case r: RuntimeReplaceable =>
r.replacement }
+ val stillUnsupported =
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction]
||
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.Generator] ||
+ !unwrapped.deterministic ||
+ !unwrapped.resolved ||
+ !unwrapped.checkInputDataTypes().isSuccess
+ if (stillUnsupported) {
+ unresolvedFunc
+ } else {
+ unwrapped
+ }
+ } match {
+ case Success(resolved) => resolved
+ case Failure(_) => unresolvedFunc
Review Comment:
**minor:** `Failure(_)` discards Spark's own diagnostic. `substring(name)`
gets `WRONG_NUM_ARGS.WITHOUT_SUGGESTION: The 'substring' requires ...` from
`lookupFunction`, but the user only sees the generic `Unsupported functions:
substring`. Relatedly, a registry function over an unknown one
(`concat(no_such_fn(name), 'x')`) is reported as `Unsupported functions:
concat, no_such_fn`, while the hardcoded `upper(no_such_fn(name))` lists only
the inner name. Not blocking. Could we keep the caught `AnalysisException`
message and surface it in the rejection text, given #19850's aim was actionable
errors?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +354,64 @@ 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(name, 'x') =
'a1x'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(name, 'a') =
1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "if(name = 'a1', true,
false)", scalarSchema))
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))
+ assertResult(Right(()))(validate("concat(name, 'x') = 'a1x'"))
+ assertResult(Right(()))(validate("instr(name, 'a') = 1"))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "nvl(name, 'z') =
'a1'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "left(name, 1) = 'a'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "right(name, 1) =
'1'", scalarSchema))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2) =
'1'", scalarSchema))
+
+ // 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))
+ }
+
+ test("evaluateFilter still rejects aggregate/generator/nondeterministic
functions resolved via FunctionRegistry") {
+ // percentile/any_value etc. 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).
+ assert(validate("any_value(id) = 1").isLeft)
Review Comment:
**minor:** This duplicates the identical assertion at line 596, and it does
not exercise the new guard: the parser lowers `any_value(...)` straight to an
`AggregateExpression`, so it never enters `resolveViaFunctionRegistry` and is
rejected by the `Unevaluable` collector on the base too (same message,
`Unsupported filter expression: aggregateexpression`). Not blocking. Could we
drop this line and let `percentile` (or `max`, see above) carry the
aggregate-guard claim in the comment?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -555,4 +602,29 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Right(()))(validate("upper(name) = 'A1'"))
}
+
+ test("evaluateFilter resolves a registry function nested inside another") {
+ // Function resolution runs bottom-up, so a registry-resolved argument
(upper(name)) is
+ // already a real expression by the time its enclosing call (instr/concat)
is checked -
+ // otherwise the outer call would look unresolved and get rejected even
though both
+ // functions individually resolve fine.
+ assertResult(Right(()))(validate("instr(upper(name), 'A') = 1"))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(upper(name),
'A') = 1", scalarSchema))
+ assertResult(Right(()))(validate("concat(upper(name), 'x') = 'A1x'"))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(upper(name),
'x') = 'A1x'", scalarSchema))
+ }
+
+ test("evaluateFilter resolves deeper nesting and more RuntimeReplaceable
functions") {
+ // Two levels of registry-only nesting.
+ assertResult(Right(()))(validate("instr(concat(name, 'x'), 'a') = 1"))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(concat(name,
'x'), 'a') = 1", scalarSchema))
+ // Registry function nested inside a hardcoded-table function, and vice
versa three levels deep.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "upper(concat(name,
'x')) = 'A1X'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(
+ keep(scalarRows, "upper(concat(lower(name), 'x')) = 'A1X'",
scalarSchema))
+ // Other RuntimeReplaceable builtins beyond nvl/left/right also need the
replacement unwrap.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "ifnull(name, 'z') =
'a1'", scalarSchema))
+ assertResult(scalarRows)(keep(scalarRows, "nvl2(name, 'yes', 'no') =
'yes'", scalarSchema))
Review Comment:
**nit:** Both rows have a non-null `name`, so `nvl2` never takes its null
branch and the expected set is the whole input; the same goes for `ifnull`
above it. Feel free to ignore. Could we run these two against the null fixture
at line 491 (`Row(2, null)`), so `nvl2(...) = 'yes'` keeps one row and
`ifnull(name, 'z') = 'z'` keeps the other?
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +401,47 @@ 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;
+ // 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; and a type
mismatch the analyzer's
+ // implicit-cast pass would normally have caught (e.g. concat on a
non-string column) still
+ // fails checkInputDataTypes. 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 or db-qualified builtins. A
3+ part name
+ // (catalog.db.func) isn't safe to look up: FunctionIdentifier only
carries one qualifier,
+ // and guessing by dropping the extra parts risks matching an unrelated
same-named function.
+ val functionIdentifier = unresolvedFunc.nameParts match {
+ case Seq(funcName) => Some(FunctionIdentifier(funcName))
+ case Seq(db, funcName) => Some(FunctionIdentifier(funcName, Some(db)))
+ case _ => None
+ }
+ val resolved = functionIdentifier
+ .map(sparkSession.sessionState.functionRegistry.lookupFunction(_,
unresolvedFunc.arguments))
+ .getOrElse(unresolvedFunc)
+ val unwrapped = resolved.transformUp { case r: RuntimeReplaceable =>
r.replacement }
+ val stillUnsupported =
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction]
||
+
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.Generator] ||
Review Comment:
**minor:** `Unevaluable` is not one of the guards, so ten builtins
(`current_user`, `current_database`, `current_catalog`, `current_timezone`,
`grouping`, `grouping_id`, `lag`, `lead`, ...) pass this method and are only
stopped by the separate `Unevaluable` collector in `validateFilterExpression`;
`evaluateFilter` on its own would drop every row. Not blocking, since all 17
in-repo procedures validate first. Could we add it as a sixth disjunct so this
method is safe by itself?
```suggestion
unwrapped.isInstanceOf[org.apache.spark.sql.catalyst.expressions.Generator] ||
unwrapped.exists(_.isInstanceOf[Unevaluable]) ||
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +354,64 @@ 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(name, 'x') =
'a1x'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(name, 'a') =
1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "if(name = 'a1', true,
false)", scalarSchema))
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))
+ assertResult(Right(()))(validate("concat(name, 'x') = 'a1x'"))
+ assertResult(Right(()))(validate("instr(name, 'a') = 1"))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "nvl(name, 'z') =
'a1'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "left(name, 1) = 'a'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "right(name, 1) =
'1'", scalarSchema))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2) =
'1'", scalarSchema))
Review Comment:
**nit:** Only `substring` covers the arity fall-through, and only in the
direction where the registry rescues the call. The other direction,
`lookupFunction` throwing `WRONG_NUM_ARGS` into the `case Failure(_)` arm, is
reached by no unit test (`no_such_fn` in `TestShowCleansProcedures` is the only
in-repo coverage). Feel free to ignore. Could we pin that arm here?
```suggestion
assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2)
= '1'", scalarSchema))
assert(validate("length(name, 1) = 2").isLeft)
```
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -555,4 +602,29 @@ class TestHoodieProcedureFilterUtils extends
HoodieSparkProcedureTestBase {
assertResult(Right(()))(validate("upper(name) = 'A1'"))
}
+
+ test("evaluateFilter resolves a registry function nested inside another") {
Review Comment:
**minor:** Both shapes here (`instr(upper(name), 'A')`, `concat(upper(name),
'x')`) are registry-over-hardcoded, and the test added one commit later at line
617 already covers that plus registry-over-registry and three levels deep
(`upper(concat(lower(name), 'x'))`). Not blocking. Could we fold this test into
that one, keeping `instr(upper(name), 'A')` as the regression case for the
`transformUp` fix and dropping the `concat(upper(name), 'x')` pair?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +354,64 @@ 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(name, 'x') =
'a1x'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(name, 'a') =
1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "if(name = 'a1', true,
false)", scalarSchema))
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))
+ assertResult(Right(()))(validate("concat(name, 'x') = 'a1x'"))
+ assertResult(Right(()))(validate("instr(name, 'a') = 1"))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "nvl(name, 'z') =
'a1'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "left(name, 1) = 'a'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "right(name, 1) =
'1'", scalarSchema))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2) =
'1'", scalarSchema))
+
+ // 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))
+ }
+
+ test("evaluateFilter still rejects aggregate/generator/nondeterministic
functions resolved via FunctionRegistry") {
+ // percentile/any_value etc. 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).
+ assert(validate("any_value(id) = 1").isLeft)
+ assert(validate("percentile(id, 0.5) = 1").isLeft)
+ assert(validate("explode(array(1, 2)) = 1").isLeft)
+ assert(validate("rand() = 1").isLeft)
+ assert(validate("uuid() = 'x'").isLeft)
+ assertResult(Seq.empty)(keep(scalarRows, "any_value(id) = 1",
scalarSchema))
+ 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_date/current_timestamp are deterministic-at-eval-time (Spark
computes them
+ // directly rather than requiring rule substitution), so they resolve and
evaluate for real
Review Comment:
**minor:** The comment covers `current_date` too, but `current_date()` is
rejected: `CurrentDate` is `TimeZoneAwareExpression`, so without a session zone
it is unresolved (`Unsupported functions: current_date`), the same limitation
as `hour(t)` at line 466. Only `current_timestamp` resolves. Not blocking.
Could we name only `current_timestamp` here and add
`assert(validate("current_date() > d").isLeft)` next to it so the difference is
pinned?
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/procedure/TestHoodieProcedureFilterUtils.scala:
##########
@@ -354,15 +354,64 @@ 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "concat(name, 'x') =
'a1x'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "instr(name, 'a') =
1", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "if(name = 'a1', true,
false)", scalarSchema))
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))
+ assertResult(Right(()))(validate("concat(name, 'x') = 'a1x'"))
+ assertResult(Right(()))(validate("instr(name, 'a') = 1"))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "nvl(name, 'z') =
'a1'", scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "left(name, 1) = 'a'",
scalarSchema))
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "right(name, 1) =
'1'", scalarSchema))
+
+ // 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.
+ assertResult(Seq(scalarRows.head))(keep(scalarRows, "substring(name, 2) =
'1'", scalarSchema))
+
+ // 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))
+ }
+
+ test("evaluateFilter still rejects aggregate/generator/nondeterministic
functions resolved via FunctionRegistry") {
+ // percentile/any_value etc. 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).
+ assert(validate("any_value(id) = 1").isLeft)
+ assert(validate("percentile(id, 0.5) = 1").isLeft)
+ assert(validate("explode(array(1, 2)) = 1").isLeft)
+ assert(validate("rand() = 1").isLeft)
+ assert(validate("uuid() = 'x'").isLeft)
+ assertResult(Seq.empty)(keep(scalarRows, "any_value(id) = 1",
scalarSchema))
+ 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_date/current_timestamp are deterministic-at-eval-time (Spark
computes them
+ // directly rather than requiring rule substitution), so they resolve and
evaluate for real
+ // instead of needing denylist treatment.
+ assertResult(scalarRows)(keep(scalarRows, "current_timestamp() > t",
scalarSchema))
Review Comment:
**nit:** `current_timestamp() > t` keeps both rows and will for any
timestamp after 2024, so it cannot fail even if the function evaluated to
garbage. Feel free to ignore. Could we pair it with the opposite direction?
```suggestion
assertResult(scalarRows)(keep(scalarRows, "current_timestamp() > t",
scalarSchema))
assertResult(Seq.empty)(keep(scalarRows, "current_timestamp() < t",
scalarSchema))
```
--
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]
