You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "LuciferYang (via GitHub)" <gi...@apache.org> on 2026/09/01 09:03:17 UTC
[PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on code in PR #58445:
URL: https://github.com/apache/spark/pull/58445#discussion_r3916899194
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1259,7 +1259,16 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan with CodegenSup
override def supportsRowBased: Boolean = children.forall(_.supportsRowBased)
protected override def doExecuteColumnar(): RDD[ColumnarBatch] = {
- sparkContext.union(children.map(_.executeColumnar()))
+ // Same split as `doExecute`: a union that reports an index-co-locatable partitioning has to
+ // interleave same-index partitions, or a parent that skipped an exchange on that report reads
+ // a concatenation instead.
+ outputPartitioning match {
Review Comment:
Done in 215ef11d49a and 8eae58a878b. The helper takes the execution action rather than the already-executed RDDs:
```scala
private def unionRDDs[T: ClassTag](executeChild: SparkPlan => RDD[T]): RDD[T] = {
val partitioning = outputPartitioning
val rdds = children.map(executeChild)
partitioning match { ... }
}
protected override def doExecute(): RDD[InternalRow] = unionRDDs(_.execute())
protected override def doExecuteColumnar(): RDD[ColumnarBatch] = unionRDDs(_.executeColumnar())
```
One difference from the snippet worth flagging: a strict `Seq[RDD[T]]` parameter evaluates `children.map(_.execute())` before the match, so `outputPartitioning` ends up being read after the children have run, which is not what `doExecute` did before. A child's partitioning can sharpen once it has executed, because `InMemoryTableScanExec.cachedPlan` unwraps an inner `AdaptiveSparkPlanExec` only while `isFinalPlan`, and materializing the cache is what sets that. Reading the partitioning into a val first keeps the arm and its `numPartitions` on one read, and it also removes the second read the old co-located arm did after executing the children.
`DataFrameSetOperationsSuite`, `UnionCodegenSuite`, `AdaptiveQueryExecSuite`, `KeyGroupedPartitioningSuite`, `CoalesceShufflePartitionsSuite` and `BucketedReadWithoutHiveSupportSuite` run 400 tests, all passing. The new test now pins its expected rows instead of re-running the query with `spark.sql.unionOutputPartitioning=false`, and I confirmed it fails without the fix.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5520793124
> Could you make backporting PRs for branch-4.2 and branch-4.1 in order to make it sure to pass the CIs, please, @LuciferYang ?
ok
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun closed pull request #58445: [SPARK-59141][SQL] Interleave partitions in columnar UnionExec
URL: https://github.com/apache/spark/pull/58445
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5519705535
> +1, LGTM. Thank you, @LuciferYang. The shared `unionRDDs` helper and the point about reading `outputPartitioning` before the children execute both look right to me.
>
> One small request before merging: could you refresh the PR description to match the latest commits? The "What changes" section does not mention the shared helper, and "How was this patch tested" still says the new test compares against the same query with `spark.sql.unionOutputPartitioning` off, while the test now pins its expected rows.
The PR description has been updated.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5520784583
Could you make backporting PRs for branch-4.2 and branch-4.1 in order to make it sure to pass the CIs, please, @LuciferYang ?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5520763563
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/f04a15f617c25733ddfdc3a216651568ed5c087f
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "LuciferYang (via GitHub)" <gi...@apache.org>.
LuciferYang commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5509338755
cc @cloud-fan @dongjoon-hyun
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5494650425
Thank you @LuciferYang! The fix is minimal and correct. It mirrors doExecute's partitioning-aware branch exactly, uses SQLPartitioningAwareUnionRDD[T: ClassTag] generically (no new RDD type needed), and avoids the double-read of outputPartitioning that the sibling PR has to patch in the row path. The test anchors the invariants that make the fix meaningful (columnar-only path taken, co-locatable partitioning reported, exchange dropped) and compares against the result from the non-partitioning-aware path. No API, config, version, or MiMa concerns.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #58445:
URL: https://github.com/apache/spark/pull/58445#issuecomment-5520774709
**Merge Summary:**
- merged into branch-4.x https://github.com/apache/spark/commit/60ad0620b0f48bee8ea08b137817c69aa955a5d0
- merged into branch-4.3 https://github.com/apache/spark/commit/d5fbdc173bd6934a50002b15e89a4a420e387eab
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59141][SQL] Interleave partitions in columnar UnionExec [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on code in PR #58445:
URL: https://github.com/apache/spark/pull/58445#discussion_r3914843635
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/basicPhysicalOperators.scala:
##########
@@ -1259,7 +1259,16 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan with CodegenSup
override def supportsRowBased: Boolean = children.forall(_.supportsRowBased)
protected override def doExecuteColumnar(): RDD[ColumnarBatch] = {
- sparkContext.union(children.map(_.executeColumnar()))
+ // Same split as `doExecute`: a union that reports an index-co-locatable partitioning has to
+ // interleave same-index partitions, or a parent that skipped an exchange on that report reads
+ // a concatenation instead.
+ outputPartitioning match {
Review Comment:
This is a copy of `doExecute`'s dispatch with `execute()` swapped for `executeColumnar()`, and the bug this PR fixes is exactly the two copies drifting: SPARK-52921 added the split only to `doExecute` while `doExecuteColumnar` already existed, and SPARK-57881 added the `KeyedPartitioning` arm only there. Since both `SQLPartitioningAwareUnionRDD` and `SparkContext.union` are generic over `T: ClassTag`, how about one helper shared by both paths so they cannot drift again?
```scala
private def unionRDDs[T: ClassTag](rdds: Seq[RDD[T]]): RDD[T] = outputPartitioning match {
case _: UnknownPartitioning | _: KeyedPartitioning =>
// (existing comment from doExecute)
sparkContext.union(rdds)
case partitioning =>
// (existing comment from doExecute)
new SQLPartitioningAwareUnionRDD(
sparkContext, rdds.filter(!_.partitions.isEmpty), partitioning.numPartitions)
}
protected override def doExecute(): RDD[InternalRow] = unionRDDs(children.map(_.execute()))
protected override def doExecuteColumnar(): RDD[ColumnarBatch] =
unionRDDs(children.map(_.executeColumnar()))
```
It needs `import scala.reflect.ClassTag`, and the "Same split as `doExecute`" comment can go away since there is only one split left.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org