Anil K created SPARK-60522:
------------------------------
Summary: Storage-partitioned join fails with INTERNAL_ERROR when a
partition transform is not a ScalarFunction
Key: SPARK-60522
URL: https://issues.apache.org/jira/browse/SPARK-60522
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Anil K
A storage-partitioned join fails with an {{INTERNAL_ERROR}} instead of falling
back to a shuffle when a connector's partition transform binds to a
{{BoundFunction}} that is not a {{ScalarFunction}}. {{V2ExpressionUtils}} keeps
whatever {{UnboundFunction.bind}} returns, but {{TransformExpression}} can only
evaluate a {{ScalarFunction}}, and two SPJ paths evaluate the transform without
checking.
Repro: the catalog's {{bucket}} binds to
{code:scala}
object NonScalarBucketFunction extends BoundFunction {
override def inputTypes(): Array[DataType] = Array(IntegerType, LongType)
override def resultType(): DataType = IntegerType
override def name(): String = "bucket"
}
{code}
{code:sql}
-- bucket4 PARTITIONED BY (bucket(4, id)), ident PARTITIONED BY (id), plain not
partitioned
-- each holds the values 0 to 7
SELECT k.id FROM bucket4 k JOIN ident i ON k.id = i.id -- (1)
SELECT k.id FROM bucket4 k JOIN plain p ON k.id = p.b -- (2)
{code}
* (1) with {{spark.sql.sources.v2.bucketing.pushPartValues.enabled}} and
{{spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled}} fails at
planning with {{[INTERNAL_ERROR] Cannot evaluate expression:
transformexpression(bucket, input[0, bigint, true], Some(4))}}, from
{{TransformExpression.eval}} via {{IdentityReducer.reduce}} and
{{EnsureRequirements.checkKeyGroupCompatible}}.
* (2) with {{spark.sql.sources.v2.bucketing.shuffle.enabled}} fails in a task
with the same error, from {{ShuffleExchangeExec.prepareShuffleDependency}}, and
the job aborts.
Both queries should return 8 rows. They do with default settings, and with a
{{ScalarFunction}} bucket.
Why:
* {{TransformExpression.resolvedFunction}} is {{None}} for a function that is
not a {{ScalarFunction}}, and {{eval}} then throws
{{cannotEvaluateExpressionError}}.
* For (1), {{KeyedShuffleSpec.isExpressionCompatible}} admits an
identity-vs-transform pair on {{allowReduce && canReduceKeys}} alone, and
{{reducersBothWays}} builds an {{IdentityReducer}} that evaluates the transform
on the identity side's partition keys.
* For (2), {{KeyedShuffleSpec.canCreatePartitioning}} accepts any transform
whose arguments are columns, and the other side is shuffled by evaluating it on
each row.
A fix should refuse such a transform in those gates, so the join shuffles both
sides.
Measured on master (fce7d51a376). (2) also applies to 4.0 and 4.1: their
{{canCreatePartitioning}} accepts any {{TransformExpression}}, and the one-side
shuffle came with SPARK-48012 (4.0.0). (1) came with SPARK-56182 (4.2.0).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
