[
https://issues.apache.org/jira/browse/SPARK-59187?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Peter Toth updated SPARK-59187:
-------------------------------
Affects Version/s: 4.2.0
4.3.0
Description:
A storage-partitioned join between two keyed sides whose struct partition keys
have differently named fields fails at planning, and a union over such keys
miscounts them. Both are the same cause: a partition key row compares its data
types exactly, so two rows of one value do not match when the columns they came
from were named differently.
h3. The failing join
{code:sql}
-- s1(id struct<a:int>, v string) partitioned by identity(id), keys
named_struct('a',1), ('a',2)
-- s2(k struct<b:int>, w string) partitioned by identity(k), keys
named_struct('b',1), ('b',2)
SELECT s1.v, s2.w FROM s1 JOIN s2 ON s1.id = s2.k
{code}
with {{spark.sql.sources.v2.bucketing.enabled}} and
{{spark.sql.sources.v2.bucketing.pushPartValues.enabled}} on. The join is
legal: {{BinaryComparison.sameType}} is {{DataType.equalsStructurally(_, _,
ignoreNullability = true)}}, so no {{Cast}} is inserted, both sides are keyed,
and the key values match. It throws:
{noformat}
[STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES] Storage-partition join
partition transforms
produced incompatible reduced types, left reducers: [] returned: ["STRUCT<a:
INT>"],
right reducers: [] returned: ["STRUCT<b: INT>"]
{noformat}
for a join that reduced nothing. The path:
# {{InternalRowComparableWrapper.equals}} compares its {{dataTypes}} before its
values, so the two sides' key rows never match.
# {{KeyedShuffleSpec.isCompatibleWith}} therefore answers false and the
co-partitioned fast path is out.
# {{EnsureRequirements}} falls into the push-common-partition-values branch,
where a pair of attributes has no reducer, so each side's reduced types are its
own and the comparison throws.
h3. The union miscount
{{KeyedPartitioning.concat}} puts the children's key rows in one list and asks
whether any repeats. Rows of two namings never match, so a union of two
children that hold one key under two namings reports unique keys when they are
not, and a consumer reads that as needing no regroup.
h3. Fix
Erase the naming where key rows are built. {{InternalRowComparableWrapper}}'s
factory builds every row at the given types with struct field names and every
nullability erased, and nothing else touched: a collation, a decimal precision
and a {{char}} length all still tell two rows apart. Nothing that compares or
hashes a row reads a field name, so this cannot move a row or change a sort
order.
h3. Affected versions
The reduced-types comparison came with SPARK-56046, released in 4.2.0, so the
failing join affects 4.2.0 and later. The union path
({{KeyedPartitioning.concat}}) exists from 4.3.0.
h3. Note
This ticket was originally filed as an Improvement, to carry the key data types
on {{KeyedPartitioning}} instead of sampling them from the first key row. That
refactor is now SPARK-59285, and this ticket keeps the user-visible defect the
work uncovered.
was:
{{KeyedPartitioning}} carries its partition keys as
{{InternalRowComparableWrapper}} rows and no types. Every reader that needs the
types samples them from the first key row, via {{keyDataTypes}}:
{code:scala}
@transient lazy val keyDataTypes: Seq[DataType] =
partitionKeys.headOption.map(_.dataTypes).getOrElse(expressionDataTypes)
{code}
Two problems follow from that.
* With no key row there is nothing to sample, so it falls back to the partition
expressions' own types. Those describe the keys only while the expressions do,
and after a join reduced both sides' keys they no longer do (SPARK-59121).
SPARK-59176 was a query that failed because one caller held that fallback
against a real answer.
* Nothing checks that the remaining key rows agree with the first one.
The types are known where the keys are built, and are already computed and
dropped there:
{code:scala}
def apply(expressions: Seq[Expression], partitionKeys: Seq[InternalRow]):
KeyedPartitioning = {
val dataTypes = expressions.map(_.dataType)
val comparableKeyWrapperFactory =
InternalRowComparableWrapper.getInternalRowComparableWrapperFactory(dataTypes)
...
{code}
The proposal is to carry them on the partitioning beside {{expressions}}, and
to have {{projectKeys}}, {{reduceKeys}},
{{KeyedShuffleSpec.createPartitioning}} and {{GroupPartitionsExec}} carry them
through the transformations they apply to the keys. Then no reader samples a
key row, and a partitioning with no key still reports what its keys would hold.
This subsumes SPARK-59176's fix, which leaves a keyless side out of one
comparison rather than giving it an answer.
Issue Type: Bug (was: Improvement)
Summary: SPJ fails on a join whose two sides name their struct
partition keys differently (was: Carry the partition key data types on
KeyedPartitioning)
> SPJ fails on a join whose two sides name their struct partition keys
> differently
> --------------------------------------------------------------------------------
>
> Key: SPARK-59187
> URL: https://issues.apache.org/jira/browse/SPARK-59187
> Project: Spark
> Issue Type: Bug
> Components: SQL
> Affects Versions: 4.2.0, 4.3.0, 5.0.0, 4.4.0
> Reporter: Peter Toth
> Priority: Major
> Labels: pull-request-available
>
> A storage-partitioned join between two keyed sides whose struct partition
> keys have differently named fields fails at planning, and a union over such
> keys miscounts them. Both are the same cause: a partition key row compares
> its data types exactly, so two rows of one value do not match when the
> columns they came from were named differently.
> h3. The failing join
> {code:sql}
> -- s1(id struct<a:int>, v string) partitioned by identity(id), keys
> named_struct('a',1), ('a',2)
> -- s2(k struct<b:int>, w string) partitioned by identity(k), keys
> named_struct('b',1), ('b',2)
> SELECT s1.v, s2.w FROM s1 JOIN s2 ON s1.id = s2.k
> {code}
> with {{spark.sql.sources.v2.bucketing.enabled}} and
> {{spark.sql.sources.v2.bucketing.pushPartValues.enabled}} on. The join is
> legal: {{BinaryComparison.sameType}} is {{DataType.equalsStructurally(_, _,
> ignoreNullability = true)}}, so no {{Cast}} is inserted, both sides are
> keyed, and the key values match. It throws:
> {noformat}
> [STORAGE_PARTITION_JOIN_INCOMPATIBLE_REDUCED_TYPES] Storage-partition join
> partition transforms
> produced incompatible reduced types, left reducers: [] returned: ["STRUCT<a:
> INT>"],
> right reducers: [] returned: ["STRUCT<b: INT>"]
> {noformat}
> for a join that reduced nothing. The path:
> # {{InternalRowComparableWrapper.equals}} compares its {{dataTypes}} before
> its values, so the two sides' key rows never match.
> # {{KeyedShuffleSpec.isCompatibleWith}} therefore answers false and the
> co-partitioned fast path is out.
> # {{EnsureRequirements}} falls into the push-common-partition-values branch,
> where a pair of attributes has no reducer, so each side's reduced types are
> its own and the comparison throws.
> h3. The union miscount
> {{KeyedPartitioning.concat}} puts the children's key rows in one list and
> asks whether any repeats. Rows of two namings never match, so a union of two
> children that hold one key under two namings reports unique keys when they
> are not, and a consumer reads that as needing no regroup.
> h3. Fix
> Erase the naming where key rows are built. {{InternalRowComparableWrapper}}'s
> factory builds every row at the given types with struct field names and every
> nullability erased, and nothing else touched: a collation, a decimal
> precision and a {{char}} length all still tell two rows apart. Nothing that
> compares or hashes a row reads a field name, so this cannot move a row or
> change a sort order.
> h3. Affected versions
> The reduced-types comparison came with SPARK-56046, released in 4.2.0, so the
> failing join affects 4.2.0 and later. The union path
> ({{KeyedPartitioning.concat}}) exists from 4.3.0.
> h3. Note
> This ticket was originally filed as an Improvement, to carry the key data
> types on {{KeyedPartitioning}} instead of sampling them from the first key
> row. That refactor is now SPARK-59285, and this ticket keeps the user-visible
> defect the work uncovered.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
