metanil commented on code in PR #59760:
URL: https://github.com/apache/spark/pull/59760#discussion_r4433919500
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -167,30 +175,153 @@ case class TransformExpression(
copy(children = newChildren)
/**
- * The scalar function call this transform stands for, with the bucket count
prepended as a
- * literal argument. `None` when the bound function is not a
[[ScalarFunction]], and also when a
- * join reduced this expression's keys, since then the call no longer
computes them. Evaluating it
- * to place a row would send the row to a partition it does not belong to.
That second arm is a
- * local gate. No caller reaches it today, because every consumer of a
reduced partitioning
- * refuses it first, and the write path never sees one.
+ * The bound function, when Spark could call it for this transform: a
[[ScalarFunction]], over
+ * keys no join reduced. See [[resolvedFunction]].
*/
- lazy val resolvedFunction: Option[Expression] = function match {
- case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
- val arguments = numBucketsOpt.fold(children)(n => Literal(n) +: children)
- Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc, arguments))
+ private def scalarFunction: Option[ScalarFunction[_]] = function match {
+ case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
Some(scalarFunc)
case _ => None
}
- override def eval(input: InternalRow): Any = resolvedFunction match {
+ /** The arguments of the call: the bucket count, if any, as a literal, then
the children. */
+ private def callArguments: Seq[Expression] =
+ numBucketsOpt.fold(children)(n => Literal(n) +: children)
+
+ /**
+ * The scalar function call this transform stands for, over its arguments as
they are. `None`
+ * when the bound function is not a [[ScalarFunction]], and also when a join
reduced this
+ * expression's keys, since then the call no longer computes them.
Evaluating it to place a row
+ * would send the row to a partition it does not belong to. That second arm
is a local gate. No
+ * caller reaches it today, because every consumer of a reduced partitioning
refuses it first,
+ * and the write path never sees one. The write path
(`DistributionAndOrderingUtils`) casts the
+ * arguments to the function's input types itself. `eval` uses
[[evaluableFunction]] instead.
+ */
+ lazy val resolvedFunction: Option[Expression] =
+ scalarFunction.map(V2ExpressionUtils.resolveScalarFunction(_,
callArguments))
+
+ /**
+ * The call `eval` runs: [[resolvedFunction]], with each argument at the
type the function
+ * declares for it in `inputTypes()`. Nothing casts the arguments after
analysis, so the call
+ * gets the values as the argument holds them. Most arguments go as they
are, when they are held
+ * the way the declared type is, which is what a function read before this
check. One kind is
+ * cast: an integral argument narrower than declared is widened with a
[[Cast]], which keeps its
+ * value and needs no time zone. A function bound to a `SMALLINT` column but
declaring `INT`
+ * stays evaluable that way. Any other mismatch, or a different number of
arguments, gives
+ * `None`, as does everything `resolvedFunction` is `None` for. See
`atDeclaredType`. A mismatch
+ * is logged at info level once per function class and argument types, since
it makes a
+ * storage-partitioned join shuffle where it would not otherwise.
+ *
+ * Resolving the call can throw, e.g. when Spark finds no method to call on
a `ScalarFunction`
+ * that inherits `produceResult` from a base class, or when connector code
it calls throws. That
+ * gives `None` too, and `eval` rethrows it. So does a `LinkageError`, e.g.
a connector class
+ * whose dependency is missing, which the planner would otherwise hit for a
layout it does not
+ * pick. The planner asks this of such layouts on every query, so a failure
is logged as a
+ * warning once per function class and kind of failure, and with its stack
trace at debug level.
+ * Not serialized, so an executor resolves it again rather than receiving
the call (or the
+ * failure) the driver resolved while planning.
+ */
+ def evaluableFunction: Option[Expression] =
evaluableFunctionAttempt.toOption.flatten
Review Comment:
Thanks, done as you suggested.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
