dongjoon-hyun commented on code in PR #58097:
URL: https://github.com/apache/spark/pull/58097#discussion_r3976797848
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -1570,7 +1570,18 @@ class Dataset[T] private[sql](
/** @inheritdoc */
def toLocalIterator(): java.util.Iterator[T] = {
- withAction("toLocalIterator", queryExecution) { plan =>
+ // This action submits one job per output partition. A channel cannot
retain its output
+ // between those jobs, so use a separate regular plan even if
queryExecution already ran.
+ // Clone the session to keep the setting off during lazy planning and AQE
replanning without
+ // changing the caller's session or its other actions.
+ val iteratorSession = SparkSession.getOrCloneSessionWithConfigsOff(
Review Comment:
**PySpark `DataFrame.rdd` / `DataFrame.toLocalIterator()` bypass both new
guards.**
This re-plan and the `DeserializeToObjectExec` gate in
`PipelinedShuffleEligibility.hasUnsupportedBoundary` only cover the JVM API.
PySpark's `DataFrame.rdd` and `DataFrame.toLocalIterator()`
(`python/pyspark/sql/classic/dataframe.py`) go through `Dataset.javaToPython` /
`Dataset.toPythonIterator`, which call `queryExecution.toRdd` on the *original*
plan: no `DeserializeToObjectExec` is inserted, and this method is never
entered, so the pipelined `ShuffledRowRDD` is handed straight to the RDD API.
With both opt-in configs on a local PySpark session:
```python
df = spark.range(0, 2000000, 1, 4).repartition(8)
df.rdd.coalesce(2).count() # deadlock: one task drains
several reduce partitions
df.rdd.map(lambda r: (r.id % 2, 1)).reduceByKey(lambda a, b: a +
b).collect() # pipelined-below-regular rejection
list(df.toLocalIterator()) # producer re-runs once per
partition
list(df.toLocalIterator(prefetchPartitions=True)) # job i+1 is submitted
while job i owns the producer -> PIPELINED_SHUFFLE_CROSS_JOB_REUSE
```
`PythonRDD.toLocalIteratorAndServe` submits one job per partition and, with
prefetch, submits the next one before the current finishes, so the last case
fails on the second partition rather than merely recomputing. SQL scripting's
`OpenCursorExec` (`executedPlan.executeToIterator()`) has the same exposure.
Since every RDD-side consumer ultimately goes through `QueryExecution.toRdd`
/ `executedPlan.execute()`, gating there (or in `javaToPython` /
`toPythonIterator`, which are the two Python entry points) would close the
Python paths and the other JVM exposures (`mapGroups(...).rdd`,
`checkpoint(eager = false)`, `LogicalRDD` inputs) at once, instead of
enumerating plan nodes. At minimum the PR description's "`Dataset.rdd` ... use
regular shuffles" and "`toLocalIterator` uses a separate regular-shuffle plan"
should say JVM-only, and a PySpark test for `df.rdd` / `df.toLocalIterator()`
would be worth adding.
--
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]
