Yuming Wang created SPARK-58120:
-----------------------------------
Summary: 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
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: [email protected]
For additional commands, e-mail: [email protected]
