feat: disable Comet by default when CometShuffleManager is not registered (#4328)

Author: andygrove

spark/src/main/scala/org/apache/comet/CometSparkSessionExtensions.scala

@@ -127,6 +127,16 @@ object CometSparkSessionExtensions extends Logging {
       return false
     }
 
+    if (COMET_EXEC_SHUFFLE_ENABLED.get(conf) && !isCometShuffleManagerEnabled(conf)) {
+      logWarning(
+        "Comet extension is disabled because spark.shuffle.manager is not set to " +
+          "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager. " +
+          "Comet provides limited benefit without its shuffle manager. " +
+          s"Set ${COMET_EXEC_SHUFFLE_ENABLED.key}=false to keep Comet enabled with " +
+          "Spark's default shuffle manager.")
+      return false
+    }
+
     // We don't support INT96 timestamps written by Apache Impala in a different timezone yet
     if (conf.getConf(SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION)) {
       logWarning(

spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala

@@ -29,6 +29,9 @@ class CometSparkSessionExtensionsSuite extends CometTestBase {
 
   test("isCometLoaded") {
     val conf = new SQLConf
+    // Disable Comet shuffle so this test can focus on other checks without needing
+    // spark.shuffle.manager to be set.
+    conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "false")
 
     conf.setConfString(CometConf.COMET_ENABLED.key, "false")
     assert(!isCometLoaded(conf))
@@ -50,6 +53,25 @@ class CometSparkSessionExtensionsSuite extends CometTestBase {
     NativeBase.setLoaded(true)
   }
 
+  test("isCometLoaded requires CometShuffleManager when shuffle.enabled=true") {
+    val conf = new SQLConf
+    conf.setConfString(CometConf.COMET_ENABLED.key, "true")
+
+    // Default: shuffle.enabled=true. Without spark.shuffle.manager set, Comet must be disabled.
+    assert(!isCometLoaded(conf))
+
+    // Opt out: shuffle.enabled=false. Comet should load (assumes native lib is available).
+    conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "false")
+    assert(isCometLoaded(conf))
+
+    // shuffle.enabled=true with the Comet shuffle manager registered: Comet should load.
+    conf.setConfString(CometConf.COMET_EXEC_SHUFFLE_ENABLED.key, "true")
+    conf.setConfString(
+      "spark.shuffle.manager",
+      "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+    assert(isCometLoaded(conf))
+  }
+
   test("Arrow properties") {
     NativeBase.setLoaded(false)
     NativeBase.load()

spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala

@@ -31,6 +31,9 @@ class CometPluginsSuite extends CometTestBase {
     conf.set("spark.executor.memory", "1G")
     conf.set("spark.executor.memoryOverhead", "2G")
     conf.set("spark.plugins", "org.apache.spark.CometPlugin")
+    conf.set(
+      "spark.shuffle.manager",
+      "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
     conf.set("spark.comet.enabled", "true")
     conf.set("spark.comet.exec.enabled", "true")
     conf.set("spark.comet.exec.onHeap.enabled", "true")