[SPARK-59141][SQL] Columnar UnionExec ignores outputPartitioning and returns wrong results - #58473
Conversation
…returns wrong results ### What changes were proposed in this pull request? `UnionExec.doExecuteColumnar` now gets the same partitioning split that `doExecute` already has: a union whose `outputPartitioning` is index-co-locatable builds a `SQLPartitioningAwareUnionRDD` that interleaves same-index partitions across children, and only an `UnknownPartitioning` or `KeyedPartitioning` union concatenates. `SQLPartitioningAwareUnionRDD` is generic over its element type, so it works for `RDD[ColumnarBatch]` with no new RDD type. ### Why are the changes needed? Wrong results on the default configuration. `doExecuteColumnar` concatenated its children unconditionally while `outputPartitioning` kept reporting the partitioning the children agreed on, so a parent that dropped its exchange on the strength of that report read a concatenation instead of an interleaving. A union of two columnar-only bucketed scans reports `HashPartitioning(k, 4)`, which lets a downstream `GROUP BY k` skip its shuffle; the concatenated columnar union then puts each key in two partitions, so the shuffle-free aggregate emits each group twice. The row path has been partitioning-aware since SPARK-52921 (`spark.sql.unionOutputPartitioning`, on by default), so this reaches back to 4.1.0. ### Does this PR introduce _any_ user-facing change? Yes, it fixes wrong results. Any union of columnar children that reports an index-co-locatable partitioning was affected. ### How was this patch tested? A new case in `DataFrameSetOperationsSuite` builds a union of two bucketed (columnar-only) scans feeding a shuffle-free aggregate, asserts the union is columnar-only, reports a `HashPartitioning`, and has no exchange, then compares against the same query with `spark.sql.unionOutputPartitioning` off. It fails with ten rows where five are expected without the fix and passes with it. `DataFrameSetOperationsSuite`, `UnionCodegenSuite`, `BucketedReadWithoutHiveSupportSuite` and `KeyGroupedPartitioningSuite` pass.
|
In fact, I already submitted the fix PR yesterday. |
uros-b
left a comment
uros-b
left a comment
There was a problem hiding this comment.
The approach seems good overall, left one comment below. After addressing, please ping @cloud-fan who has more context in SQL physical operators / UnionExec area, for further review @hemanthboyina!
| outputPartitioning match { | ||
| case _: UnknownPartitioning | _: KeyedPartitioning => | ||
| sparkContext.union(children.map(_.executeColumnar())) | ||
| case _ => |
There was a problem hiding this comment.
outputPartitioning is called twice: once as the match scrutinee and once inside the arm for .numPartitions. The two calls recompute from children on every invocation, so if a child's reported partitioning changes between them (the precise scenario documented by PR #58419 for InMemoryTableScanExec under AQE - UnknownPartitioning(0) before isFinalPlan, then HashPartitioning after), the first call could route to case _ while the second call returns numPartitions = 0, constructing a SQLPartitioningAwareUnionRDD with zero partitions and producing an empty result. PR #58419 applies the same single-capture fix to doExecute (val partitioning = outputPartitioning; partitioning match { ... SQLPartitioningAwareUnionRDD(sc, rdds, partitioning.numPartitions) }). The new doExecuteColumnar should use the same pattern for consistency and to stay correct once #58419 lands.
|
my mistake @LuciferYang haven't observed your PR, closing mine |
|
Thank you @hemanthboyina |
|
oh sorry I haven't seen @LuciferYang's comment above! |
3 participants
What changes were proposed in this pull request?
UnionExec.doExecuteColumnar now gets the same partitioning split that doExecute already has: a union whose outputPartitioning is index-co-locatable builds a SQLPartitioningAwareUnionRDD that interleaves same-index partitions across children, and only an UnknownPartitioning or KeyedPartitioning union concatenates. SQLPartitioningAwareUnionRDD is generic over its element type, so it works for
RDD[ColumnarBatch]with no new RDD type.Why are the changes needed?
Wrong results on the default configuration.
doExecuteColumnarconcatenated its children unconditionally whileoutputPartitioningkept reporting the partitioning the children agreed on, so a parent that dropped its exchange on the strength of that report read a concatenation instead of an interleaving.A union of two columnar-only bucketed scans reports
HashPartitioning(k, 4), which lets a downstreamGROUP BY kskip its shuffle; the concatenated columnar union then puts each key in two partitions, so the shuffle-free aggregate emits each group twice.Does this PR introduce any user-facing change?
Yes, it fixes wrong results. Any union of columnar children that reports an index-co-locatable partitioning was affected.
How was this patch tested?
A new case in
DataFrameSetOperationsSuitebuilds a union of two bucketed (columnar-only) scans feeding a shuffle-free aggregate, asserts the union is columnar-only, reports aHashPartitioning, and has no exchange, then compares against the same query withspark.sql.unionOutputPartitioningoff. It fails with ten rows where five are expected without the fix and passes with it.DataFrameSetOperationsSuite,UnionCodegenSuite,BucketedReadWithoutHiveSupportSuiteandKeyGroupedPartitioningSuitepass.Was this patch authored or co-authored using generative AI tooling?
Yes, used Claude code