voonhous commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3964798933
##########
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:
Addressed at 97c30d35dfd7, and better than proposed: `applyImplicitCasts`
runs the analyzer's `ImplicitTypeCasts` on the raw lookup result before
`checkInputDataTypes`, so `split_part(action, 1, 1) = 'clean'` now keeps both
rows like Spark instead of being rejected, while `split_part(name, array(1),
1)` still fails the wrapper's contract. Verified both on the head.
##########
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:
Addressed at 97c30d35dfd7: `unwrapReplacements` recurses until `fastEquals`,
and `regexp_substr(action, 'lean') = 'lean'` on the `show_cleans` schema now
validates and keeps both rows, matching Spark. Pinned by the new fixed-point
test.
##########
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:
Addressed at 97c30d35dfd7: `SparkThrowable` and `IllegalArgumentException`
are rethrown, so `to_number(action, '999') > 0`, `bit_get(version, 99) = 0` and
`regexp_replace(action, '[', 'x')` now surface as `Failed to parse or evaluate
filter expression` instead of an empty result, pinned by the new runtime-error
test. Residual, fine to leave: `xpath_string` throws a plain `RuntimeException`
and is still swallowed.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
Review Comment:
Addressed at 97c30d35dfd7: `applyCoercionRules` now runs over the unwrapped
registry result before the guard. On the `show_cleans` schema
`nvl(time_taken_in_millis, 0) > 100`, `if(time_taken_in_millis > 100, true,
false)` and `nullif(time_taken_in_millis, 150) is null` each keep the one row
Spark keeps, pinned by the new nvl-vs-coalesce test. `greatest` still needs
`FunctionArgumentConversion`, which the newer thread at line 452 covers.
##########
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:
Addressed at 97c30d35dfd7: `max(id) > 0` and `explode(...)` assert their
messages, and re-running the mutation confirms it: with the `AggregateFunction`
clause deleted `max(version) > 0` validates, and with the `Generator` clause
deleted `explode` loses the `Unsupported functions` message the test now
requires. The redundant `checkInputDataTypes` clause stays, which is harmless.
##########
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:
Addressed at 97c30d35dfd7: `unwrapped.exists(_.isInstanceOf[Unevaluable])`
is in the guard. `current_user() = 'x'` and `lag(version) = 1` are rejected as
`Unsupported functions: ...`, and `evaluateFilter` on its own returns no rows
without throwing.
##########
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:
Addressed at 97c30d35dfd7: the 2-part arm is gone, `default.upper(name)`
falls into the same unresolved path as the 3-part case and is pinned next to it.
##########
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:
Addressed at 97c30d35dfd7: `assertKeeps` pairs `validate` with `keep` for
concat, instr, nvl, left, right, substring and the OR case, and the multi-name
message is pinned with `no_such_fn`/`other_missing`. `if(...)` and `case when`
stay keep-only, but both already validate, so not worth another round.
--
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]
