hudi-agent commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3981289837
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +419,157 @@ 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
+ }
+ } match {
+ case Success(resolved) => resolved
+ case Failure(_) => unresolvedFunc
+ }
+ }
+
+ // Runs a handful of the analyzer's own coercion rules on a single
expression, the same rules
+ // lookupFunction skips: ImplicitTypeCasts for nodes declaring a real
input-type contract,
+ // FunctionArgumentConversion/ConcatCoercion/IfCoercion for the builtins
whose argument types get
+ // unified before the type check even sees them (concat(id, 'x') casts the
Int to String via
+ // ConcatCoercion in a real query; without it Concat.checkInputDataTypes
just fails). Order
+ // mirrors TypeCoercion's own rule list - ImplicitTypeCasts last, as a
catch-all. Anything not
+ // covered by one of these four passes through unchanged.
+ private def applyImplicitCasts(expression: Expression): Expression = {
+ val engine: TypeCoercionBase = if (SQLConf.get.ansiEnabled)
AnsiTypeCoercion else TypeCoercion
+ Seq(engine.FunctionArgumentConversion, engine.ConcatCoercion,
engine.IfCoercion, engine.ImplicitTypeCasts)
+ .foldLeft(expression) { (expr, rule) => rule.transform.applyOrElse(expr,
identity[Expression]) }
+ }
+
+ // Filter expressions only ever call plain builtins. 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.
+ //
+ // Each argument gets widened before the lookup, not just the call's own
result afterward: an
+ // argument like ts + 1 (Long + Int) is still an unresolved Add at this
point, and the wrapper's
+ // checkInputDataTypes right after runs before pass three ever gets a chance
to widen it -
+ // sqrt(ts + 1) would fail that check for the same reason nvl(ts, 0) needed
pre-lookup widening,
+ // while the hardcoded abs(ts + 1) already works because pass three widens
its argument too, just
+ // later in the pipeline.
+ private def lookupBuiltin(unresolvedFunc: UnresolvedFunction, sparkSession:
SparkSession): Expression =
+ unresolvedFunc.nameParts match {
+ case Seq(funcName) =>
+ val widenedArguments = unresolvedFunc.arguments.map(applyCoercionRules)
+ sparkSession.sessionState.functionRegistry
+ .lookupFunction(builtinFunctionIdentifier(funcName),
widenedArguments)
+ case _ => unresolvedFunc
+ }
+
+ // A session's function registry is a clone of FunctionRegistry.builtin,
keyed the same way
+ // builtins are actually registered - a bare name pre-4.2, but the fully
qualified
+ // system.builtin.<name> from 4.2 onward, where a session-level clone
(unlike the builtin
+ // singleton itself) stops auto-qualifying a bare name it's given and
asserts instead.
+ private def builtinFunctionIdentifier(funcName: String): FunctionIdentifier =
+ if (HoodieSparkUtils.gteqSpark4_2) {
+ FunctionIdentifier(funcName, Some("builtin"), Some("system"))
Review Comment:
🤖 I think this won't compile under `-Pspark3.3`: `FunctionIdentifier` only
gained the `catalog` parameter in 3.4 (3.3.4's companion has just
`apply(String)` and `apply(String, Option[String])`), and `hudi-spark` is
shared across all Spark profiles. CI can't catch it since the spark3.3 matrix
entry is commented out. Could this go through reflection (like the `With`
handling below) or a SparkAdapter hook instead of a direct 3-arg call?
<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:
##########
@@ -62,13 +66,18 @@ object HoodieProcedureFilterUtils {
// Binding and resolution depend only on the schema, so run the three
passes once for the
// whole batch instead of per row.
- val boundExpr = bindAndResolveExpression(parsedExpr, schema)
+ val boundExpr = bindAndResolveExpression(parsedExpr, schema,
sparkSession)
rows.filter(row => evaluateExpressionOnRow(boundExpr, row, schema))
} match {
case Success(filteredRows) => filteredRows
// Surface an overflowing ANSI cast or arithmetic, or an ANSI cast of
a malformed string,
// with Spark's own exception rather than restating it as a
filter-expression problem: the
- // expression is fine, the data does not fit.
+ // expression is fine, the data does not fit. A per-row runtime error
from a
Review Comment:
🤖 nit: the added sentences here explain why a case was *not* added — that
reads more like review discussion than code doc. Could you trim it to one line
(e.g. "registry-function runtime errors are rethrown in evaluateExpressionOnRow
and land in the generic Failure branch below") and leave the rest to the PR
description?
<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 +419,157 @@ 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
+ }
+ } match {
+ case Success(resolved) => resolved
+ case Failure(_) => unresolvedFunc
+ }
+ }
+
+ // Runs a handful of the analyzer's own coercion rules on a single
expression, the same rules
+ // lookupFunction skips: ImplicitTypeCasts for nodes declaring a real
input-type contract,
+ // FunctionArgumentConversion/ConcatCoercion/IfCoercion for the builtins
whose argument types get
+ // unified before the type check even sees them (concat(id, 'x') casts the
Int to String via
+ // ConcatCoercion in a real query; without it Concat.checkInputDataTypes
just fails). Order
+ // mirrors TypeCoercion's own rule list - ImplicitTypeCasts last, as a
catch-all. Anything not
+ // covered by one of these four passes through unchanged.
+ private def applyImplicitCasts(expression: Expression): Expression = {
Review Comment:
🤖 nit: `applyImplicitCasts` and `applyCoercionRules` are easy to mix up, and
this one runs more than ImplicitTypeCasts
(FunctionArgumentConversion/ConcatCoercion/IfCoercion too). Could you rename it
to something like `applySparkTypeCoercionRules` so it's clear this is Spark's
analyzer rules vs. our own numeric widening in `applyCoercionRules`?
<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]
