metanil opened a new pull request, #59760: URL: https://github.com/apache/spark/pull/59760
### What changes were proposed in this pull request? A storage-partitioned join no longer evaluates a partition transform that Spark cannot evaluate. It shuffles both sides instead. `KeyedShuffleSpec` asks this in the two places that admit such an evaluation: - `isExpressionCompatible` admits an identity side against a transform only if Spark can evaluate the transform. `reducersBothWays` builds an `IdentityReducer` that evaluates the transform on the identity side's partition keys, and it only runs on a pair `areKeysCompatible` admitted, so it needs no change. - `canCreatePartitioning` admits a transform as the layout other children are shuffled onto only if Spark can evaluate it, since the children shuffled onto it are routed by evaluating it on each row. Both call a new private helper, `canEvaluate`, which is `TransformExpression.resolvedFunction.isDefined`. Resolving a `ScalarFunction` throws `SCALAR_FUNCTION_NOT_FULLY_IMPLEMENTED` when Spark finds no method to call on it, e.g. one that inherits `produceResult` from a base class. `canEvaluate` counts that as cannot evaluate. The gates are asked of every candidate, so throwing there would fail planning for a candidate that is never chosen. ### Why are the changes needed? `TransformExpression` can evaluate only a bound function that is a `ScalarFunction`, and `eval` throws for any other. A connector can report such a transform, since `V2ExpressionUtils` keeps whatever `UnboundFunction.bind` returns. Take a catalog whose `bucket` binds to a plain `BoundFunction`, `bucket4` partitioned by `bucket(4, id)`, `ident` partitioned by `id`, and `plain` not partitioned, each holding 0 to 7: ```sql 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) ``` - (1) with `spark.sql.sources.v2.bucketing.pushPartValues.enabled` and `spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled` fails while planning with `[INTERNAL_ERROR] Cannot evaluate expression: transformexpression(bucket, input[0, bigint, true], Some(4))`. `isExpressionCompatible` admits the pair on the confs alone, and `IdentityReducer` evaluates `bucket` on `ident`'s partition keys. - (2) with `spark.sql.sources.v2.bucketing.shuffle.enabled` fails in a task with the same error. `canCreatePartitioning` admits `bucket(4, id)` as the layout, and `ShuffleExchangeExec` evaluates it on every row of `plain`. Both should return 8 rows, which they do with the default confs, or with a `bucket` that is a `ScalarFunction`. A `ScalarFunction` that Spark cannot resolve fails the identity reduce the same way, with `SCALAR_FUNCTION_NOT_FULLY_IMPLEMENTED`. The one-side shuffle of a transform came with SPARK-48012, in 4.0.0, and the identity reduce with SPARK-56182, in 4.2.0. #58943 (SPARK-59688) already noted that the identity reduce can raise `cannotEvaluateExpressionError` for a transform bound to something that is not a `ScalarFunction`. ### Does this PR introduce _any_ user-facing change? Yes, it fixes the failures above. Such a join now shuffles both sides and returns its rows. A join whose transforms Spark can evaluate plans as before. No configuration or API changes. ### How was this patch tested? New tests: - `ShuffleSpecSuite`, "SPARK-60522: KeyedShuffleSpec refuses a transform Spark cannot evaluate". For a scalar `bucket`, a `BoundFunction` that is not a `ScalarFunction`, and a `ScalarFunction` that inherits `produceResult`, it checks `areKeysCompatible` against an identity side in both directions, and `canCreatePartitioning`. They hold only for the scalar one, and the gates do not throw for the inherited one. It fails on master. - `KeyGroupedPartitioningSuite`, "SPARK-60522: a transform Spark cannot evaluate is not reduced onto or shuffled onto". It runs (1) and (2) for each of the three functions, and checks the rows and the shuffles: none and one for the scalar `bucket`, two and two for the others. It fails on master with the `INTERNAL_ERROR`. - `KeyGroupedPartitioningSuite`, "SPARK-60522: a child Spark cannot evaluate leaves the other child as the shuffle target". `bucket(4, id)` over a function that inherits `produceResult` is joined with `bucket(8, id)` in a second catalog, whose `bucket` is a `ScalarFunction`, under `spark.sql.sources.v2.bucketing.shuffle.enabled`. The join shuffles `bucket4` onto `bucket(8, id)`, in both join orders. It passes on master, which never evaluates `bucket4`'s function here. It fails if `canEvaluate` throws instead of returning false, since `EnsureRequirements` asks `canCreatePartitioning` of both children before it picks one. Also ran `ShuffleSpecSuite`, `EnsureRequirementsSuite` and the `KeyGroupedPartitioning*` suites, 372 tests in all, plus `dev/lint-scala`. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5.5) -- 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]
