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: [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]
