LuciferYang opened a new pull request, #58445:
URL: https://github.com/apache/spark/pull/58445
### What changes were proposed in this pull request?
`UnionExec.doExecuteColumnar` 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, and only an `UnknownPartitioning` or `KeyedPartitioning`
union concatenates. `SQLPartitioningAwareUnionRDD` is generic over its element
type, so no new RDD is needed.
### Why are the changes needed?
Wrong results on default configuration. `doExecuteColumnar` concatenated its
children unconditionally, while `outputPartitioning` went on reporting whatever
the children agreed on, so a parent that dropped its exchange on the strength
of that report read a concatenation instead of an interleaving.
```scala
spark.conf.set("spark.sql.sources.bucketing.autoBucketedScanEnabled",
"false")
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.range(0, 20, 1, 1).selectExpr("id % 5 AS k").write.bucketBy(4,
"k").saveAsTable("t1")
spark.range(20, 40, 1, 1).selectExpr("id % 5 AS k").write.bucketBy(4,
"k").saveAsTable("t2")
sql("SELECT k, count(*) c FROM (SELECT k FROM t1 UNION ALL SELECT k FROM t2)
GROUP BY k").show()
```
Five groups of eight come back as ten rows of four, each group reported
twice. The plan shows why:
```
*(1) HashAggregate(keys=[k#8L], functions=[count(1)])
+- *(1) ColumnarToRow
+- Union
:- FileScan parquet t1[k#8L] Batched: true, Bucketed: true,
SelectedBucketsCount: 4 out of 4
+- FileScan parquet t2[k#10L] Batched: true, Bucketed: true,
SelectedBucketsCount: 4 out of 4
```
There is no exchange under the aggregate, because the union reported
`hashpartitioning(k#8L, 4)` from its two bucketed children.
`FileSourceScanExec` overrides `supportsColumnar` but not `supportsRowBased`,
whose default is its negation, so a batch-readable bucketed scan is
columnar-only and so is a union of two of them. `ColumnarToRowExec` therefore
goes above the union rather than below it, the union runs `doExecuteColumnar`,
and the aggregate reads partitions in which a key can appear twice.
The row path has been partitioning-aware since SPARK-52921, which is where
`spark.sql.unionOutputPartitioning` came from, so this reaches back to 4.1.0.
Fixing the other side, by reporting `UnknownPartitioning` whenever the union
might run columnar, does not work as well. `EnsureRequirements` reads
`outputPartitioning` at `QueryExecution.preparations`, well before
`ApplyColumnarRulesAndInsertTransitions` decides anything, so at that point the
only available proxy is `supportsColumnar`. That is true for plenty of unions
that end up on the row path, and using it would give up SPARK-52921's exchange
elimination for all of them.
### Does this PR introduce _any_ user-facing change?
Yes. The query above returns five rows instead of ten. Any union of
columnar-only children that reports an index-co-locatable partitioning was
affected the same way.
### How was this patch tested?
A new case in `DataFrameSetOperationsSuite`, alongside the existing
SPARK-52921 union partitioning cases. It builds the shape above and asserts
that the union is columnar-only, that it reports a `HashPartitioning`, and that
the aggregate has no exchange, before comparing against the same query with
`spark.sql.unionOutputPartitioning` off. Without the fix it fails with ten rows
where five are expected.
`DataFrameSetOperationsSuite`, `BucketedReadWithoutHiveSupportSuite` and
`UnionCodegenSuite` run 127 cases, and `AdaptiveQueryExecSuite`,
`KeyGroupedPartitioningSuite` and `CoalesceShufflePartitionsSuite` run 273, all
passing. `sql/scalastyle` and `sql/Test/scalastyle` are clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
