anshulsingh-py commented on code in PR #19853:
URL: https://github.com/apache/hudi/pull/19853#discussion_r3991828384
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -357,10 +366,28 @@ object HoodieProcedureFilterUtils {
}
case _ => unresolvedFunc
}
+ resolveOrFallback(hardcodedResolved, unresolvedFunc, sparkSession)
}
// Third pass: handle type coercion for numeric comparisons
- functionResolved.transformUp {
+ applyCoercionRules(functionResolved)
+ }
+
+ // Whatever the hardcoded table produced - a real expression, or nothing at
all (wrong arity, or
+ // a name not in the table) - still an UnresolvedFunction? Try the registry
before giving up.
+ private def resolveOrFallback(firstAttempt: Expression, original:
UnresolvedFunction, sparkSession: SparkSession): Expression =
+ firstAttempt match {
+ case _: UnresolvedFunction => resolveViaFunctionRegistry(original,
sparkSession)
+ case resolved => resolved
+ }
+
+ // Widens numeric comparison/arithmetic operands to a common type - see the
coercion helpers
+ // below for the rules each case follows. Shared between the third pass here
and
+ // resolveViaFunctionRegistry, so a RuntimeReplaceable unwrap (nvl ->
Coalesce, for instance)
+ // gets the same widening its hardcoded-table equivalent (coalesce) gets,
before either is
+ // checked for remaining type errors.
+ private def applyCoercionRules(expression: Expression): Expression = {
Review Comment:
Renamed to `applyHudiWideningRules` (this file's own numeric widening) and
`applySparkAnalyzerCoercionRules` (Spark's real analyzer rules) - keeps
`applyTypeCoercion` distinct as the single-node helper the widening pass calls.
##########
hudi-spark-datasource/hudi-spark/src/main/scala/org/apache/spark/sql/hudi/command/procedures/HoodieProcedureFilterUtils.scala:
##########
@@ -389,6 +416,166 @@ 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 =
applySparkTypeCoercionRules(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 applySparkTypeCoercionRules(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
= {
+ val bareIdentifier = FunctionIdentifier(funcName)
+ if (HoodieSparkUtils.gteqSpark4_2) {
+ // FunctionIdentifier only gained the catalog parameter from Spark 3.4
onward - this file
+ // still compiles against 3.3 too, where the case class has just
funcName/database, so a
+ // direct 3-arg call wouldn't compile there. Reached through reflection
instead, the same way
+ // the With handling below reaches classes that don't exist on every
targeted version.
+ bareIdentifier.getClass
+ .getConstructor(classOf[String], classOf[Option[_]],
classOf[Option[_]])
+ .newInstance(funcName, Some("builtin"), Some("system"))
+ .asInstanceOf[FunctionIdentifier]
+ } else {
+ bareIdentifier
+ }
+ }
+
+ // RuntimeReplaceable placeholders (nvl, ifnull, left, right, ...) need
substitution the analyzer
+ // normally performs but lookupFunction skips, and can themselves unwrap to
another
+ // RuntimeReplaceable (regexp_substr -> NullIf), so the unwrap runs to a
fixed point. Then widens
+ // numeric operands the same way pass three would - nvl(ts, 0) unwraps to
Coalesce(ts, 0), which
+ // needs the same widening the hardcoded coalesce(ts, 0) case gets - so a
registry function and
+ // its hardcoded-table equivalent agree on what counts as resolved.
+ //
+ // A replacement can itself be a With(child, defs) common-subexpression
wrapper from 4.0 onward
+ // (NullIf's, for instance) - Unevaluable like any other holder, so it needs
inlining here too or
+ // isUsableOutsideQueryPlan rejects it outright.
With/CommonExpressionDef/CommonExpressionRef
+ // don't exist before 4.0, so this goes by reflection rather than a direct
import; evaluating a
+ // filter once per row rather than once per query makes the dedup With
exists for irrelevant, so
+ // inlining each reference in place of its definition is exactly equivalent
to keeping it.
+ private def finalizeRegistryResolution(expression: Expression): Expression =
{
+ def unwrapReplacements(expr: Expression): Expression = {
+ val next = expr.transformUp {
+ case r: RuntimeReplaceable => r.replacement
+ case withExpr if isWithNode(withExpr) =>
inlineCommonExpressions(withExpr)
+ }
+ if (next.fastEquals(expr)) next else unwrapReplacements(next)
+ }
+ applyCoercionRules(unwrapReplacements(expression))
+ }
+
+ // With/CommonExpressionDef/CommonExpressionRef don't exist before Spark
4.0, so these go by
+ // reflection rather than a direct import to keep this file compiling across
the same 3.3-4.2
+ // range as the rest of it - the single place to update if the class name or
shape ever changes.
+ private def isWithNode(expression: Expression): Boolean =
expression.getClass.getSimpleName == "With"
+
+ private def isCommonExpressionRef(expression: Expression): Boolean =
+ expression.getClass.getSimpleName == "CommonExpressionRef"
+
+ private def inlineCommonExpressions(withExpr: Expression): Expression = {
+ val defsById = withExpr.getClass.getMethod("defs").invoke(withExpr)
+ .asInstanceOf[Seq[Expression]]
+ .map { commonExprDef =>
+ val id = commonExprDef.getClass.getMethod("id").invoke(commonExprDef)
+ val child =
commonExprDef.getClass.getMethod("child").invoke(commonExprDef).asInstanceOf[Expression]
+ id -> child
+ }.toMap
+ val child =
withExpr.getClass.getMethod("child").invoke(withExpr).asInstanceOf[Expression]
+ child.transformUp {
+ case ref if isCommonExpressionRef(ref) =>
+ defsById(ref.getClass.getMethod("id").invoke(ref))
+ }
+ }
+
+ // A resolved expression still isn't usable one row at a time if it's an
aggregate (percentile,
+ // collect_list - only make sense across real aggregation), a generator
(explode, inline - only
+ // work inside a projection), still Unevaluable somewhere in it
(current_user, lag, lead, ... -
+ // only valid in their normal analyzer context), or non-deterministic (rand,
uuid,
+ // spark_partition_id - expect per-partition initialization this evaluator
never does).
+ //
+ // Unevaluable alone doesn't cover the whole family on every Spark version:
current_timestamp
+ // and its relatives are foldable (Spark computes them once and reuses the
value, rather than
+ // per row) but which marker trait exempts them from eval() varies even
within the 4.x line this
+ // builds against (current_timestamp is Unevaluable-free on 4.0 but
foldable-but-uneval'able
+ // there, fine again on 4.1+), so a foldable result gets a real probe
instead of a trait check -
+ // eval() against EmptyRow only touches its own constant inputs, never a
real column, so a throw
+ // here means it genuinely can't be evaluated standalone rather than that
this row's data is
+ // missing.
+ //
+ // Only a bare SparkException - the marker Unevaluable.eval() itself throws
- counts as that
+ // structural "can't evaluate at all" signal. An all-literal call can also
be foldable (nothing
+ // references a column), and a data-specific failure there
(regexp_replace('a', '[', 'x'), an
+ // invalid pattern) is a SparkThrowable or IllegalArgumentException, not a
bare SparkException -
+ // treating it as unusable here would silently reject it instead of letting
it raise normally,
+ // exactly the silent-drop behavior the whole registry fallback exists to
avoid.
+ private def isUsableOutsideQueryPlan(expression: Expression): Boolean = {
Review Comment:
Trimmed to the one non-obvious rule. The per-version current_timestamp
history was already duplicated next to the test that pins it, so it only needed
to live there.
--
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]
