HyukjinKwon commented on code in PR #58808:
URL: https://github.com/apache/spark/pull/58808#discussion_r4014298333
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -2132,7 +2132,10 @@ class Dataset[T] private[sql](
* Converts a JavaRDD to a PythonRDD.
*/
private[sql] def javaToPython: JavaRDD[Array[Byte]] = {
- javaToPython(queryExecution.withRegularShuffle)
+ val qe = queryExecution.withRegularShuffle
+ withNewRDDExecutionId("javaToPython", qe) {
Review Comment:
Thanks. This is the pre-existing behavior of the classic `Dataset.rdd` path
(SPARK-50994) that this PR mirrors for PySpark. `rddQueryExecution` there is
also built via `withRegularShuffle` and inherits the same `RemoveShuffleFiles`
cleanup mode, so a later action on the returned RDD (e.g.
`dataset.rdd.collect()`) recomputes outside the execution in exactly the same
way; the classic SPARK-50994 test only avoids it by re-wrapping into a
DataFrame + `checkAnswer` (a fresh execution) rather than acting on the RDD
directly.
Under the default production config
(`spark.sql.classic.shuffleDependency.fileCleanup.enabled=false`, since it
defaults to `Utils.isTesting`) the eagerly-materialized shuffle is retained and
reused by the deferred action, so the propagation holds. The recompute only
happens when cleanup is enabled, which is the tradeoff that conf documents:
"this cleanup may cause stage retries and regenerate shuffle files if the same
dataframe reference is executed again."
Closing this gap for real means using `DoNotCleanup` for RDD-conversion
executions, and to stay consistent it should cover both the classic and PySpark
paths, which is broader than this parity port. I would rather keep this PR
aligned with the classic behavior and address the deeper cleanup interaction
separately.
##########
sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala:
##########
@@ -2132,7 +2132,10 @@ class Dataset[T] private[sql](
* Converts a JavaRDD to a PythonRDD.
*/
private[sql] def javaToPython: JavaRDD[Array[Byte]] = {
- javaToPython(queryExecution.withRegularShuffle)
+ val qe = queryExecution.withRegularShuffle
+ withNewRDDExecutionId("javaToPython", qe) {
Review Comment:
This is likewise identical to the classic `Dataset.rdd` (SPARK-50994): it
also wraps only the lazy RDD construction in `withNewRDDExecutionId`, so
`SparkListenerSQLExecutionEnd` and `observationManager.tryComplete` fire when
the lineage is built, before any action on the returned RDD. AQE eagerly
materializes the shuffle stages within that window, but the final stage is
deferred, so the terminal state is published early on both paths.
Since the goal here is parity with SPARK-50994, I have kept the same
semantics rather than diverging on the PySpark side. Making
listeners/observations span the deferred RDD action is a design change that
should be done for the classic and PySpark paths together, so I would prefer to
handle it as separate follow-up rather than in this parity port.
--
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]
