voonhous commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3964799857
##########
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:
Addressed at 97c30d35dfd7: `concat(action, 'x') = 'cleanx'` added to
`filterTests`; the suite passes locally on this head.
##########
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:
Addressed at 97c30d35dfd7: the comment names only `current_timestamp`, and
`current_date() > d` is pinned as rejected.
##########
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:
Addressed at 97c30d35dfd7: the duplicate is dropped and `max(id)` carries
the aggregate-guard claim.
##########
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:
Addressed at 97c30d35dfd7: folded into the deeper-nesting test, keeping
`instr(upper(name), 'A')` as the regression case.
--
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]
