[
https://issues.apache.org/jira/browse/SPARK-60522?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Anil K updated SPARK-60522:
---------------------------
Affects Version/s: 4.0.4
4.1.3
4.2.0
(was: 5.0.0)
> 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: 4.2.0, 4.1.3, 4.0.4
> Reporter: Anil K
> Priority: Major
>
> 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]
