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: [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]
