peter-toth commented on code in PR #59760:
URL: https://github.com/apache/spark/pull/59760#discussion_r4435873395
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala:
##########
@@ -356,27 +356,57 @@ case class EnsureRequirements(
// consider the minimum parallelism iff ALL children need to be
re-shuffled.
//
// A child needs to be re-shuffled iff either of the following is true:
- // 1. It can't create partitioning by itself, i.e.,
`canCreatePartitioning` returns false
- // (as for the case of `RangePartitioning`), therefore it needs to be
re-shuffled
- // according to another child's shuffle spec.
+ // 1. It can't create partitioning by itself, i.e., no member of its
spec can serve as the
+ // layout (`canServe` below, as for the case of `RangePartitioning`),
therefore it needs
+ // to be re-shuffled according to another child's shuffle spec.
// 2. It already has `ShuffleExchangeLike`, so we can re-use existing
shuffle without
// introducing extra shuffle.
//
// On the other hand, in scenarios such as:
// HashPartitioning(5) <-> HashPartitioning(6)
// while `spark.sql.shuffle.partitions` is 10, we'll only re-shuffle the
left side and make it
// HashPartitioning(6).
- val shouldConsiderMinParallelism = children.zip(specs).forall { case
(child, spec) =>
- spec.forall(!_.canCreatePartitioning) ||
child.isInstanceOf[ShuffleExchangeLike]
+ //
+ // A member of a child serves as the layout when it can create a
partitioning, and either Spark
+ // can evaluate that partitioning or no child has to be shuffled onto it:
every other
+ // co-partitioned child pairs with it as it stands, which evaluates
nothing. One that cannot
+ // serve does not keep its child from being re-shuffled, so it counts
towards neither the
+ // parallelism guard nor the ranking below.
+ //
+ // A `ShuffleSpecCollection` answers `isCompatibleWith` if *any* of its
members does, so the
+ // winner alone does not say which member the sides agreed on. The
projection pushed into a
+ // paired child and the partitioning built for a re-shuffled child both
have to come from one
+ // member, otherwise the sides end up grouped on different keys, or on a
key set the child does
+ // not even have. So what is picked here is the pairing, not a member per
side.
+ //
+ // Per child, each member of its spec with the member each child pairs
with it, in child order.
+ // The pairing compares partition keys, so it is worked out when first
asked for, and once: for
+ // a member that cannot be evaluated by `canServe`, and for the winner's
members below.
+ val childMembers: Seq[Seq[LeafShuffleSpec]] =
specs.map(_.toSeq.flatMap(_.flatten))
+ final class Pairing(val member: LeafShuffleSpec, index: Int) {
+ lazy val paired: Seq[Option[LeafShuffleSpec]] =
+ childMembers.map(_.find(member.isCompatibleWith))
+ // `canEvaluatePartitioning` first: it is kept by the spec, while the
pairing compares keys.
+ lazy val canServe: Boolean =
Review Comment:
**Finding 7.** The new `EnsureRequirementsSuite` tests give each child one
member. So `canServe`, and the member filter after the ranking, are not run for
a child whose `PartitioningCollection` holds both a member Spark can evaluate
and one it cannot. Two cases would pin the choice:
- the other child pairs only with the member Spark cannot evaluate: the pair
is read as it stands;
- the other child pairs with neither: it is shuffled onto the member Spark
can evaluate.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -202,4 +324,62 @@ object TransformExpression {
case t: TransformExpression => t.reducedWith.isDefined
case _ => false
}
+
+ /**
+ * `argument` at `declaredType`, or `None` when the function could not take
it. See
+ * [[TransformExpression.evaluableFunction]].
+ *
+ * An argument goes as it is when it is stored the way the declared type is
(`storedAlike`),
+ * which is how a function read it before this check. A numeric argument
narrower than declared
+ * (`UpCastRule.legalNumericPrecedence`), e.g. a `SMALLINT` against an `INT`
or an `INT` against a
+ * `DOUBLE`, is held as another JVM type, so it is cast, which gives the
value a magic `invoke`
+ * got from Java's widening before this check, and needs no time zone.
Anything else is refused,
+ * a `DATE` against a `TIMESTAMP` in particular, whose cast takes the
session time zone.
+ */
+ private def atDeclaredType(argument: Expression, declaredType: DataType):
Option[Expression] =
+ (argument.dataType, declaredType) match {
+ case (argumentType, _) if storedAlike(argumentType, declaredType, nested
= false) =>
+ Some(argument)
+ case (argumentType, _) if
UpCastRule.legalNumericPrecedence(argumentType, declaredType) =>
+ Some(Cast(argument, declaredType))
+ case _ =>
+ None
+ }
+
+ /**
+ * Whether a value of type `from` is stored the way one of type `to` is, so
that a function
+ * declaring `to` reads it as it is:
+ * - two types with one physical type, e.g. a `VARCHAR` and a string of the
same collation, a
+ * `DATE` and an `INT`, or a `TIMESTAMP` and a `BIGINT`. A user-defined
type is stored as its
+ * `sqlType`;
+ * - two structs, arrays or maps whose fields, elements, keys and values
are stored alike, by
+ * position. Field names and nullability do not change how a value is
stored. A function
+ * declaring a non-nullable field gets the values a nullable one holds;
+ * - two decimals of one scale, so that the unscaled digits a function
reads are the same. At the
+ * top level the function gets the `Decimal` itself, whatever precision
it declares, e.g. a
+ * `DECIMAL(12,2)` key of a cogroup against a function declaring
`DECIMAL(10,2)`, which a
+ * cogroup does not coerce. Nested in an unsafe row, a decimal is read at
the declared
+ * precision: it is stored as a `long` up to `Decimal.MAX_LONG_DIGITS`
digits and as bytes
+ * above, so there `to` needs no less precision, and both on one side of
that.
+ */
+ private def storedAlike(from: DataType, to: DataType, nested: Boolean):
Boolean =
+ (from, to) match {
+ case _ if from == to => true
+ case (udt: UserDefinedType[_], _) => storedAlike(udt.sqlType, to, nested)
+ case (_, udt: UserDefinedType[_]) => storedAlike(from, udt.sqlType,
nested)
+ case (f: StructType, t: StructType) =>
+ f.length == t.length && f.fields.zip(t.fields).forall { case (ff, tf)
=>
+ storedAlike(ff.dataType, tf.dataType, nested = true)
+ }
+ case (f: ArrayType, t: ArrayType) => storedAlike(f.elementType,
t.elementType, nested = true)
+ case (f: MapType, t: MapType) =>
+ storedAlike(f.keyType, t.keyType, nested = true) &&
+ storedAlike(f.valueType, t.valueType, nested = true)
+ case (f: DecimalType, t: DecimalType) =>
+ f.scale == t.scale && (!nested || (f.precision <= t.precision &&
+ DecimalType.isByteArrayDecimalType(f) ==
DecimalType.isByteArrayDecimalType(t)))
+ case _ =>
+ val physical = PhysicalDataType(from)
+ physical != UninitializedPhysicalType && physical ==
PhysicalDataType(to)
Review Comment:
**Finding 6.** This compares physical types only, so two datetime or
interval types that share one but count in different units go as they are:
- `TIME` holds nanoseconds of the day in a `long`, where `TIMESTAMP`,
`TIMESTAMP_NTZ` and a `DAY TO SECOND` interval hold microseconds;
- a `YEAR TO MONTH` interval holds months in an `int`, where a `DATE` holds
days;
- a `DAY TO SECOND` interval holds a duration, which a function declaring
`TIMESTAMP` reads as an instant.
A connector that binds such a column to a function declaring the other type
gets the raw value in the wrong unit. That is the `years`-over-`DATE` shape
this PR fixes, and it stays as on master. This is from reading, since the test
catalog binds no such pair.
Letting two datetime or interval types pass only when they are equal would
refuse these. `TIMESTAMP_NTZ` against `TIMESTAMP` can stay, since both hold
microseconds and the connector's keys don't depend on the session time zone.
`DATE` against `INT` and `TIMESTAMP` against `BIGINT`, which the description
names, are not affected. A `ShuffleSpecSuite` row for `TIME` against
`TIMESTAMP` would pin it. If you'd rather keep this PR as it is, a ticket for
it works too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2570,17 +2605,28 @@ case class KeyedShuffleSpec(
// `createPartitioning` replaces a transform's argument with the other
child's cluster key,
// which drops whatever surrounds the key, e.g. the `+ 1` of
`bucket(4, b + 1)` or the `.a`
// of `bucket(4, s.a)`. Such a spec's own child would then be laid out
differently from the
- // children shuffled onto it.
+ // children shuffled onto it. Whether Spark can evaluate it is
`canEvaluatePartitioning`.
case t: TransformExpression => argumentsAreColumns(t)
case _ => false
} &&
// Shuffling another child onto these keys evaluates the partition
expressions per row to
// decide where each row goes, and reduced keys are not what those
expressions compute.
partitioning.expressionsDescribeKeys
+ /** Whether Spark can evaluate each transform of this layout, see
`canEvaluate`. */
+ override lazy val canEvaluatePartitioning: Boolean =
partitioning.expressions.forall {
+ case t: TransformExpression => canEvaluate(t)
+ case _ => true
+ }
+
override def createPartitioning(clustering: Seq[Expression]): Partitioning =
{
assert(clustering.size == distribution.clustering.size,
"Required distributions of join legs should be the same size.")
+ // A child shuffled onto this layout is placed by evaluating it, which
would fail on an
+ // executor. A caller has to ask `canEvaluatePartitioning` first, so a
caller that does not
+ // fails here, while planning. It is asked of this layout's own keys, see
`canEvaluate`.
+ assert(canEvaluatePartitioning,
Review Comment:
**Finding 5.** This asserts the template, this layout's own keys. A cogroup
does not coerce its keys, so the partitioning built below over a shuffled
child's key of another type can still be one Spark cannot evaluate. Keeping
that case loud is right, as
[r4433782711](https://github.com/apache/spark/pull/59760#discussion_r4433782711)
explains, since the hash fallback would pair the groups wrongly (SPARK-60782).
But it fails in a task, not here while planning as the comment says.
Measured with a `FlatMapCoGroupsInPandasExec` in `EnsureRequirementsSuite`:
the left child is `KeyedPartitioning(bucket(4, cL))` over an `INT`, with a
function declaring `(INT, INT)`, and the right key `cR` is a `BIGINT`, under
`spark.sql.sources.v2.bucketing.shuffle.enabled`:
- this head plans the shuffle onto `bucket(4, cR)`, and evaluating it fails
with `[INTERNAL_ERROR] Cannot evaluate expression: transformexpression(bucket,
input[0, bigint, true], Some(4))`;
- master fails with `[INTERNAL_ERROR] Couldn't find method invoke with
arguments (int, long)`.
Asserting what is built, after `newExpressions`, makes it fail while
planning:
```scala
assert(newExpressions.forall {
case t: TransformExpression => t.evaluableFunction.isDefined
case _ => true
}, s"Cannot shuffle a child onto a layout Spark cannot evaluate:
$newExpressions")
```
With it the probe fails here with that message. `ShuffleSpecSuite`,
`TransformExpressionSuite`, `DistributionSuite`, `KeyGroupedPartitioningSuite`,
`EnsureRequirementsSuite`, `GroupPartitionsExecSuite` and
`ValidateRequirementsSuite` still pass.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/ShuffleSpecSuite.scala:
##########
@@ -705,6 +713,245 @@ class ShuffleSpecSuite extends SparkFunSuite with
SQLHelper {
}
}
+ test("SPARK-60522: KeyedShuffleSpec refuses a transform Spark cannot
evaluate") {
+ // A connector may bind its partition transform to a `BoundFunction` that
is not a
+ // `ScalarFunction`, or to a `ScalarFunction` whose `produceResult` is
inherited, which
+ // resolving it does not find. `TransformExpression.eval` cannot run
either, so an identity
+ // side is not reduced onto it and no other child is shuffled onto it. The
gates answer that
+ // without throwing, since they are asked of candidates that may not be
chosen.
+ val nonScalarBucket = NonScalarBucketFunction
+ abstract class BucketBase extends ScalarFunction[Int] {
+ override def produceResult(input: InternalRow): Int = input.getInt(1)
+ }
+ // Declares the bucket count too, so that only the inherited
`produceResult` refuses it.
+ val inheritedBucket = new BucketBase {
+ override def inputTypes(): Array[DataType] = Array(IntegerType,
IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "bucket"
+ override def canonicalName(): String = "test.bucket"
+ }
+ val a = $"a".int
+ def spec(expression: Expression): KeyedShuffleSpec = KeyedShuffleSpec(
+ KeyedPartitioning(Seq(expression), Seq(InternalRow(1))),
ClusteredDistribution(Seq(a)))
+ val identity = spec(a)
+
+ Seq(
+ ("scalar", TestBucketFunction, true),
+ ("non-scalar", nonScalarBucket, false),
+ ("inherited produceResult", inheritedBucket, false)
+ ).foreach {
+ case (name, fn, evaluable) =>
+ val bucket = spec(TransformExpression(fn, Seq(a), Some(4)))
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_PARTIALLY_CLUSTERED_DISTRIBUTION_ENABLED.key
-> "false",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+ assert(identity.areKeysCompatible(bucket, allowReduce = true) ==
evaluable, name)
+ assert(bucket.areKeysCompatible(identity, allowReduce = true) ==
evaluable, name)
+ }
+ withSQLConf(SQLConf.V2_BUCKETING_SHUFFLE_ENABLED.key -> "true") {
+ assert(bucket.canCreatePartitioning, s"$name: a layout as it stands")
+ assert(bucket.canEvaluatePartitioning == evaluable, name)
+ }
+ }
+ }
+
+ test("SPARK-60027: KeyedShuffleSpec refuses a transform whose arguments miss
its input types") {
+ // Nothing casts a transform's arguments, the bucket count included, to
its function's
+ // `inputTypes()` after analysis, so the call would get the raw values. So
the gates require
+ // the arguments to match the input types in number and type. A narrower
integral argument is
Review Comment:
**Finding 9.** These comments still describe the rule before `43c97a16474`
and `53d75f325d6`:
- here: a narrower numeric argument is also widened to a floating-point
type, and types stored alike pass as they are. A `BIGINT` widened to `DOUBLE`
can also lose digits, as with the analyzer's implicit cast on the write path,
so "the cast keeps its value" no longer holds either;
- `ShuffleSpecSuite.scala:797`: "that can hold its value", while a top-level
decimal of the declared scale now passes at any precision;
- `KeyGroupedPartitioningSuite.scala:1894-1895`: "A narrower integral
argument is widened ... Anything else is refused", while the test widens `INT`
to `DOUBLE` and keeps `DateYearsFunction`'s plan;
- `TransformExpressionSuite.scala:151`: "not integral" for a `STRING`
argument, which is now "not numeric".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -167,30 +175,144 @@ 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]].
+ */
+ private def scalarFunction: Option[ScalarFunction[_]] = function match {
+ case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
Some(scalarFunc)
+ case _ => None
+ }
+
+ /** 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()`, see `atDeclaredType` for which
arguments suit. `None` when
+ * an argument does not, when there is another number of arguments, when the
call fails to
+ * resolve, and wherever `resolvedFunction` is `None`. `eval` rethrows a
failure to resolve. A
+ * storage-partitioned join shuffles instead of evaluating such a transform,
so each case is
+ * logged, see `logArgumentsMismatch` and `logCannotResolve`.
*/
- 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))
+ def evaluableFunction: Option[Expression] = evaluableFunctionAttempt match {
+ case Success(call) => call
case _ => None
}
- override def eval(input: InternalRow): Any = resolvedFunction match {
- case Some(fn) => fn.eval(input)
- case None => throw QueryExecutionErrors.cannotEvaluateExpressionError(this)
+ /**
+ * The call, or why it failed to resolve, e.g. a `ScalarFunction` that
inherits `produceResult`
+ * from a base class, connector code that throws, or a `LinkageError` for a
connector class whose
+ * dependency is missing, which the planner would otherwise hit for a layout
it does not pick.
+ * Computed and logged once, since a `lazy val` whose initializer throws
runs it again on each
+ * access. Not serialized, so an executor resolves it again rather than
receiving the call (or
+ * the failure) the driver resolved while planning.
+ */
+ @transient private lazy val evaluableFunctionAttempt:
Try[Option[Expression]] =
+ try Success(resolveEvaluableFunction()) catch {
+ case NonFatal(e) => cannotResolve(e)
+ case e: LinkageError => cannotResolve(e)
+ }
+
+ private def cannotResolve(e: Throwable): Try[Option[Expression]] = {
+ TransformExpression.logCannotResolve(function, e)
+ Failure(e)
+ }
+
+ private def resolveEvaluableFunction(): Option[Expression] =
+ scalarFunction.flatMap { scalarFunc =>
+ val arguments = callArguments
+ val declaredTypes = scalarFunc.inputTypes().toSeq
+ val typed = arguments.zip(declaredTypes).map { case (argument,
declaredType) =>
+ TransformExpression.atDeclaredType(argument, declaredType)
+ }
+ if (arguments.length == declaredTypes.length &&
typed.forall(_.isDefined)) {
+ Some(V2ExpressionUtils.resolveScalarFunction(scalarFunc,
typed.flatten))
+ } else {
+ TransformExpression.logArgumentsMismatch(scalarFunc,
arguments.map(_.dataType),
+ declaredTypes)
+ None
+ }
+ }
+
+ override def eval(input: InternalRow): Any = evaluableFunctionAttempt match {
+ case Success(Some(call)) => call.eval(input)
+ case Failure(e) => throw e
+ case _ => throw QueryExecutionErrors.cannotEvaluateExpressionError(this)
}
override protected def doGenCode(ctx: CodegenContext, ev: ExprCode):
ExprCode =
throw QueryExecutionErrors.cannotGenerateCodeForExpressionError(this)
}
-object TransformExpression {
+object TransformExpression extends Logging {
+ /**
+ * The function classes and kinds of failure a failure to resolve a call was
logged as a warning
+ * for, as `functionClass:failureClass`. A function class that fails in
another way is warned
+ * about again.
+ */
+ private val failuresWarnedAbout: java.util.Set[String] =
ConcurrentHashMap.newKeySet[String]()
+
+ /**
+ * The function classes and argument types a mismatch with the function's
input types was
+ * logged for, as `functionClass:argumentTypes`.
+ */
+ private val mismatchesLogged: java.util.Set[String] =
ConcurrentHashMap.newKeySet[String]()
+
+ /**
+ * `text`, or `fallback` when it throws. `name()` and a failure's message
are connector code,
+ * which can throw as the call did, e.g. a `NoClassDefFoundError` for a
missing dependency. A log
+ * line must not fail planning.
+ */
+ private def orElse(text: => String, fallback: String): String =
+ try text catch { case NonFatal(_) | _: LinkageError => fallback }
+
+ /**
+ * Logs that `function`'s input types do not take `argumentTypes`, see
`evaluableFunction`.
+ */
+ private def logArgumentsMismatch(
+ function: BoundFunction,
+ argumentTypes: Seq[DataType],
+ declaredTypes: Seq[DataType]): Unit = {
+ def typeList(types: Seq[DataType]): String =
types.map(_.catalogString).mkString("(", ", ", ")")
+ val functionClass = function.getClass.getName
+ val arguments = typeList(argumentTypes)
+ if (mismatchesLogged.add(s"$functionClass:$arguments")) {
Review Comment:
**Finding 10.** The key is the function class and the argument types, but
the message names the declared types too. A class whose instances declare other
`inputTypes()`, e.g. one bound per decimal precision or per column type, logs
only its first mismatch for the same argument types. Adding the declared types
to the key, and to the doc of `mismatchesLogged`, keeps one line per mismatch
the message describes:
```suggestion
if
(mismatchesLogged.add(s"$functionClass:$arguments:${typeList(declaredTypes)}"))
{
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -167,30 +175,144 @@ 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]].
+ */
+ private def scalarFunction: Option[ScalarFunction[_]] = function match {
+ case scalarFunc: ScalarFunction[_] if reducedWith.isEmpty =>
Some(scalarFunc)
+ case _ => None
+ }
+
+ /** 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()`, see `atDeclaredType` for which
arguments suit. `None` when
+ * an argument does not, when there is another number of arguments, when the
call fails to
+ * resolve, and wherever `resolvedFunction` is `None`. `eval` rethrows a
failure to resolve. A
+ * storage-partitioned join shuffles instead of evaluating such a transform,
so each case is
Review Comment:
**Finding 8.** `resolveEvaluableFunction` is `scalarFunction.flatMap { ...
}`, so a function that is not a `ScalarFunction` returns `None` without calling
either log helper. That is the SPARK-60522 case, so a join that loses its
reduce or one-side plan for it says nothing. Either log it at INFO like a
mismatch, or say here that only a mismatch and a failure to resolve are logged.
--
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]
