cloud-fan commented on code in PR #58808:
URL: https://github.com/apache/spark/pull/58808#discussion_r4016805475
##########
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, that makes sense. Since this PR matches the existing Dataset.rdd
lifecycle and fixes the production-default cleanup-disabled path, I agree the
cleanup-enabled retry should be handled as a separate classic-and-PySpark
follow-up.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4013687951","thread_id":"inline:4013687951","verdict_sha256":"10e1049654a095eb31793dc9333fd025fbb0704cc67914adfad97792aa860e90"}
-->
##########
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:
Agreed. This PR should preserve the SPARK-50994 execution boundary rather
than redesign listener and Observation lifetime only for PySpark; that shared
lifecycle can be considered separately.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4013687961","thread_id":"inline:4013687961","verdict_sha256":"10e1049654a095eb31793dc9333fd025fbb0704cc67914adfad97792aa860e90"}
-->
##########
python/pyspark/sql/tests/test_dataframe.py:
##########
@@ -1228,6 +1228,37 @@ def test_to_json(self):
self.assertIsInstance(df, DataFrame)
self.assertEqual(df.select("value").count(), 10)
+ def test_rdd_conversion_propagates_sql_conf(self):
+ # Converting a DataFrame to an RDD must run under a tracked SQL
execution so
+ # that session SQLConfs are propagated to the executors, mirroring the
classic
+ # Scala Dataset.rdd behavior (SPARK-50994). Otherwise the
non-vectorized parquet
+ # reader on the executor side would not see spark.sql.caseSensitive
and would
+ # resolve the two same-named but differently-cased columns wrongly,
failing to
+ # read the file. Accessing `.rdd` eagerly materializes the shuffle map
stage
+ # (the parquet scan) under adaptive execution; disabling shuffle-file
cleanup
+ # (off by default outside of tests) lets `collect` reuse that
materialized
+ # output instead of recomputing the scan without the propagated conf.
+ with self.sql_conf(
+ {
+ "spark.sql.caseSensitive": True,
+ "spark.sql.parquet.enableVectorizedReader": False,
+ "spark.sql.classic.shuffleDependency.fileCleanup.enabled":
False,
Review Comment:
Understood. The comment makes the production-default coverage and cleanup
tradeoff explicit, so I do not think this parity test should lock in the
cleanup-enabled failure mode.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4013687965","thread_id":"inline:4013687965","verdict_sha256":"10e1049654a095eb31793dc9333fd025fbb0704cc67914adfad97792aa860e90"}
-->
--
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]
