[jira] [Updated] (SPARK-58120) BatchScanExec.doCanonicalize reorders keyGroupedPartitioning expressions, causing ClassCastException in storage partitioned joins

["Yuming Wang (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/SPARK-58120?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Yuming Wang updated SPARK-58120:
--------------------------------
    Labels:   (was: v2)

> BatchScanExec.doCanonicalize reorders keyGroupedPartitioning expressions, causing ClassCastException in storage partitioned joins
> ---------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: SPARK-58120
>                 URL: https://issues.apache.org/jira/browse/SPARK-58120
>             Project: Spark
>          Issue Type: Bug
>          Components: SQL
>    Affects Versions: 4.2.0, 3.5.9, 5.0.0
>            Reporter: Yuming Wang
>            Priority: Major
>
> h2. Problem
> `BatchScanExec.doCanonicalize()` uses `QueryPlan.normalizePredicates()` to canonicalize `keyGroupedPartitioning` expressions:
> {code:scala}
> override def doCanonicalize(): BatchScanExec = {
>   this.copy(
>     output = output.map(QueryPlan.normalizeExpressions(_, output)),
>     runtimeFilters = QueryPlan.normalizePredicates(
>       runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)),
>       output),
>     keyGroupedPartitioning = keyGroupedPartitioning.map(QueryPlan.normalizePredicates(_, output)))
>   )
> }
> {code}
> `QueryPlan.normalizePredicates()` combines expressions with `And`, canonicalizes the combined expression (which can reorder children), then splits back. This destroys the original expression order.
> For example, if `keyGroupedPartitioning` is `Seq(id, data)` where `id` is `IntegerType` and `data` is `StringType`, after canonicalization the expressions may be reordered to `Seq(data, id)`. When this canonicalized `BatchScanExec` ends up in the execution plan (e.g., through a fallback or reuse path), `outputPartitioning` creates a `KeyedPartitioning` with expression data types `[StringType, IntegerType]` but partition key row values `(Integer, String)`. This type mismatch causes `ClassCastException` during `TimSort` when sorting partition keys by the expression-derived ordering.
> h2. Impact
> This bug manifests when join keys are a subset of partition keys (e.g., SPARK-44647, SPARK-47094 scenarios). The `ClassCastException` occurs in `DataSourceV2ScanExecBase.outputPartitioning()` when sorting `inputPartitions` by `partitionKey()` using `keyRowOrdering` built from the (reordered) expression data types:
> {noformat}
> java.lang.ClassCastException: class java.lang.Integer cannot be cast to class org.apache.spark.unsafe.types.UTF8String
>   at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow.getUTF8String(rows.scala:45)
>   ...
>   at org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanExecBase.outputPartitioning(DataSourceV2ScanExecBase.scala:99)
> {noformat}
> h2. Reproduction
> Create a table partitioned by two columns with different types (e.g., `identity(id)` as `IntegerType` and `identity(data)` as `StringType`). When the plan is canonicalized (which happens through exchange reuse, AQE, or Gluten's fallback mechanism), the expression order is reversed, causing the `ClassCastException`.
> h2. Fix
> Use `QueryPlan.normalizeExpressions` instead of `QueryPlan.normalizePredicates` for `keyGroupedPartitioning`. `normalizeExpressions` normalizes attribute references without combining/reordering, preserving the expression order:
> {code:scala}
> override def doCanonicalize(): BatchScanExec = {
>   this.copy(
>     output = output.map(QueryPlan.normalizeExpressions(_, output)),
>     runtimeFilters = QueryPlan.normalizePredicates(
>       runtimeFilters.filterNot(_ == DynamicPruningExpression(Literal.TrueLiteral)),
>       output),
>     keyGroupedPartitioning = keyGroupedPartitioning.map(_.map(
>       QueryPlan.normalizeExpressions(_, output))))
>   )
> }
> {code}
> h2. Affected versions
> This affects all Spark versions that have the `keyGroupedPartitioning` field in `BatchScanExec.doCanonicalize()`, including the current master branch.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org