cloud-fan commented on code in PR #58808:
URL: https://github.com/apache/spark/pull/58808#discussion_r4013687951
##########
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:
**Non-blocking (P2):** Because this execution returns before the Python RDD
is consumed, cleanup=true removes any AQE shuffle materialized while SQLConf is
propagated. The later collect can retry the scan after the local properties
have been restored, so the same case-sensitive Parquet scenario can still fail
or resolve columns incorrectly under a supported setting. The propagation
boundary needs to avoid ending a cleanup-capable SQL execution before the
returned RDD action.
##########
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:
**Non-blocking (P2):** This wrapper closes as soon as lazy RDD lineage is
built. SQLExecution then posts success and calls ObservationManager.tryComplete
before any Python RDD action runs, so listeners can report success before a
later failure and an Observation can be permanently completed with empty
metrics. RDD construction should not publish terminal query state.
##########
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:
**Non-blocking (P2):** This override disables the supported branch in which
the new execution removes the eagerly materialized shuffle before collect. The
test can therefore stay green even though cleanup=true recomputes without
propagated SQLConf and reproduces the wrong or failing Parquet read. Please
cover the cleanup-enabled path as well.
--
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]
