viirya commented on code in PR #56928:
URL: https://github.com/apache/spark/pull/56928#discussion_r3511099072
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/LocalTableScanExec.scala:
##########
@@ -55,12 +59,23 @@ case class LocalTableScanExec(
if (rows.isEmpty) {
sparkContext.emptyRDD
} else {
- val numSlices = math.min(
- unsafeRows.length, session.leafNodeDefaultParallelism)
+ val numSlices = if (useSingleTask) {
+ 1
+ } else {
+ math.min(unsafeRows.length, session.leafNodeDefaultParallelism)
+ }
sparkContext.parallelize(unsafeRows.toImmutableArraySeq, numSlices)
}
}
+ override def outputPartitioning: Partitioning = {
+ if (useSingleTask) {
+ SinglePartition
Review Comment:
Good catch — this was a real correctness issue. The advertised
`SinglePartition` did not match the zero-partition `emptyRDD`: with the shuffle
elided, a global aggregation over an empty marked relation ran zero tasks and
returned no rows instead of the single row expected on empty input. Fixed by
producing one empty partition when the scan is marked, matching how
`maybeCoalesceInputRDD` handles the all-files-pruned case in
`FileSourceScanExec`, and added a test that failed before 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]
