[I] Queries fail when `spark.shuffle.manager` names the Comet shuffle manager only in the session conf [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove opened a new issue, #6199:
URL: https://github.com/apache/datafusion-comet/issues/6199

   ### Describe the bug
   
   `CometSparkSessionExtensions.isCometShuffleManagerEnabled` decides whether the application runs Comet's shuffle manager from the class name in `spark.shuffle.manager` in the session's `SQLConf` ([CometSparkSessionExtensions.scala#L190-L197](https://github.com/apache/datafusion-comet/blob/67803a7a422c44de07af1e5d25c1dbeae8df68d4/spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala#L190-L197)). The session conf can name a manager the application is not running. The shuffle manager is created once per `SparkEnv`, but once a `SparkContext` exists, `SparkSession.builder().config(...).getOrCreate()` still copies core configs into the session's `SQLConf` without applying them. An existing session goes through `applyModifiableSettings`, which only logs "Using an existing Spark session; only runtime SQL configurations will take effect", and a new session on the running context merges them with `mergeNonStaticSQLConfigs`.
   
   When the session conf names `CometShuffleManager` but the application runs Spark's `SortShuffleManager`, `isCometLoaded` and `isCometShuffleEnabled` both pass, Comet plans a `CometExchange`, and the reduce side fails because `CometShuffledBatchRDD.createReader` casts the real manager's reader to `CometShuffleReader` ([CometShuffledRowRDD.scala#L115-L122](https://github.com/apache/datafusion-comet/blob/67803a7a422c44de07af1e5d25c1dbeae8df68d4/spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometShuffledRowRDD.scala#L115-L122)):
   
   ```
   java.lang.ClassCastException: class org.apache.spark.shuffle.BlockStoreShuffleReader cannot be cast to class org.apache.spark.sql.comet.execution.shuffle.CometShuffleReader
   	at org.apache.spark.sql.comet.execution.shuffle.CometShuffledBatchRDD.createReader(CometShuffledRowRDD.scala:121)
   	at org.apache.spark.sql.comet.execution.shuffle.CometShuffledBatchRDD.compute(CometShuffledRowRDD.scala:183)
   ```
   
   The query fails the same way with `spark.comet.shuffle.mode` set to `auto`, `native` or `jvm`.
   
   The opposite mismatch disables Comet for no reason: if the application runs `CometShuffleManager` and the session conf names another manager, `isCometLoaded` logs that `spark.shuffle.manager` is not set to Comet's manager and turns Comet off.
   
   The Celeborn path already avoids this. `cometCelebornShuffleFallbackReason` inspects the manager that `SparkEnv` holds, and `CometCelebornShufflePlanningSuite` checks that a session override cannot change the result. Only the local manager is taken from the session conf.
   
   ### Steps to reproduce
   
   Start an application with `CometSparkSessionExtensions` registered, off-heap memory enabled and `spark.comet.enabled=true`, but without setting `spark.shuffle.manager`, so Spark runs its own shuffle manager. Comet is disabled, as expected, because Comet shuffle is enabled by default and the Comet shuffle manager is not running. Then:
   
   ```scala
   val session = SparkSession.builder()
     .config("spark.shuffle.manager", "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
     .getOrCreate() // returns the existing session
   session.read.parquet(path).groupBy("key").count().collect()
   ```
   
   The plan now contains `CometExchange hashpartitioning(...), ENSURE_REQUIREMENTS, CometNativeShuffle`, and the job fails with the `ClassCastException` above.
   
   ### Expected behavior
   
   Comet decides whether it can plan Comet shuffles from the shuffle manager that `SparkEnv` holds, not from the session conf. In the example, Comet stays disabled as it was before the builder call. In the opposite case, Comet stays enabled.
   
   ### Additional context
   
   Found while addressing review feedback on #6195, where `isCometLoaded` had the same problem with `spark.memory.offHeap.enabled`. The fix will go in the same PR.
   


-- 
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: github-unsubscribe@datafusion.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [I] Queries fail when `spark.shuffle.manager` names the Comet shuffle manager only in the session conf [datafusion-comet]

["andygrove (via GitHub)" <gi...@apache.org>]

andygrove closed issue #6199: Queries fail when `spark.shuffle.manager` names the Comet shuffle manager only in the session conf
URL: https://github.com/apache/datafusion-comet/issues/6199


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org