Peter Toth created SPARK-59176:
----------------------------------
Summary: Reduced partition keys report the un-reduced data types
when the partitioning has no key
Key: SPARK-59176
URL: https://issues.apache.org/jira/browse/SPARK-59176
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 5.0.0
Reporter: Peter Toth
{{KeyedPartitioning.keyDataTypes}} falls back to the partition expressions' own
data types when the partitioning has no partition key:
{code:scala}
@transient lazy val keyDataTypes: Seq[DataType] =
partitionKeys.headOption.map(_.dataTypes).getOrElse(expressionDataTypes)
{code}
That fallback is not truthful once a storage-partitioned join has reduced both
sides' keys onto a common key space. The partitioning then reports an
expression that no longer describes its keys, so the fallback hands back the
un-reduced transform's type while the other leg of the same reduce reports the
reducer's result type.
A partitioning can reach that state with no key at all: with
{{spark.sql.sources.v2.bucketing.partition.filter.enabled}} an inner join whose
two sides hold disjoint keys intersects to none.
h3. Repro
Four tables over {{(id, ts)}}, with a {{days}} and a {{years}} function whose
reducers both produce {{LongType}}:
* {{days1}} and {{years1}} hold 2020 and 2021, so that leg keeps its keys.
* {{days2}} holds only 2020 and {{years2}} only 2021, so that leg intersects to
no key.
Join {{days1}} to {{years1}} and {{days2}} to {{years2}}, then join the two
legs on {{ts}}, with {{pushPartValues}}, {{partition.filter.enabled}},
{{allowKeysSubsetOfPartitionKeys}} and {{allowCompatibleTransforms}} all
enabled.
The correct answer is empty. Instead the query fails at planning with
{{STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES}}: the empty leg
contributes {{DateType}} from {{days(ts)}}, the other {{LongType}} from the
reducer, and {{EnsureRequirements}}' reduced-types check compares the two.
h3. Notes
The untruthful fallback arrived with SPARK-59120. The failure at the
reduced-types check becomes reachable through SPARK-59121, which stops deriving
a second reducer from an already reduced pair, so both sides read
{{keyDataTypes}} directly. Before SPARK-59121 the same query failed too, with a
{{ClassCastException}} from applying a reducer to already reduced values, so
this is not a regression in either change. Found while reviewing SPARK-59121
(apache/spark#58447) and documented in the {{keyDataTypes}} scaladoc there.
h3. Two ways to fix it
# Record the reducer's {{resultType()}} alongside the pairing in
{{TransformExpression.reducedWith}}, so {{keyDataTypes}} stays truthful with no
key to read.
# Let the reduced-types check in {{EnsureRequirements}} tolerate a side that
has no partition key, since such a side has no key type to contribute.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
