feat: warn about unrecognized spark.comet configuration keys

.github/workflows/pr_build_linux.yml

@@ -604,6 +604,7 @@ jobs:
               org.apache.comet.CometNativeSuite
               org.apache.comet.DirectColumnarToRowConverterSuite
               org.apache.comet.CometConfSuite
+              org.apache.comet.CometConfValidatorSuite
               org.apache.comet.CometPublicApiSuite
               org.apache.comet.QueryContextInternerSuite
               org.apache.comet.CometSetOpWithGroupBySuite

.github/workflows/pr_build_macos.yml

@@ -242,6 +242,7 @@ jobs:
               org.apache.comet.CometNativeSuite
               org.apache.comet.DirectColumnarToRowConverterSuite
               org.apache.comet.CometConfSuite
+              org.apache.comet.CometConfValidatorSuite
               org.apache.comet.CometPublicApiSuite
               org.apache.comet.QueryContextInternerSuite
               org.apache.comet.CometSetOpWithGroupBySuite

docs/source/contributor-guide/adding_a_new_operator.md

@@ -452,6 +452,12 @@ Support levels:
 
 Note that Comet will treat an operator as incompatible if any of the child expressions are incompatible.
 
+An operator that can return `Incompatible()` must register its
+`spark.comet.operator.<OperatorName>.allowIncompatible` key in `CometConf` with
+`createOperatorIncompatConfig` (a contrib registers a config entry for it through
+`CometConfigProvider`).
+Otherwise Comet warns that the key is not recognized when a user sets it.
+
 ### Step 3: Register the Operator
 
 Add your operator to the appropriate map in `CometExecRule.scala`:

docs/source/user-guide/latest/configs.md

@@ -21,6 +21,10 @@ under the License.
 
 Comet provides the following configuration settings.
 
+When a `spark.comet.*` key is set that Comet does not recognize, the driver logs a warning once
+for that key, and suggests the closest known key when one is close. `spark.comet.datafusion.*`
+keys are passed to DataFusion as they are and are not checked.
+
 ## Scan Configuration Settings
 
 <!--BEGIN:CONFIG_TABLE[scan]-->

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

@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import java.util.Locale
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.CometDriverPlugin
+import org.apache.spark.internal.Logging
+import org.apache.spark.unsafe.types.UTF8String
+
+import org.apache.comet.serde.QueryPlanSerde
+
+/**
+ * Warns about `spark.comet.*` keys that Comet does not recognize, which are most likely typos
+ * that Spark would otherwise accept silently.
+ *
+ * A key is recognized when it is a registered config (core or contrib) or one of its
+ * alternatives, an `enabled` or `allowIncompatible` key of an expression Comet has a serde for,
+ * or `spark.comet.version`. `spark.comet.datafusion.*` keys are passed to DataFusion as they are,
+ * so they are not checked. Operator `allowIncompatible` keys are only read for operators whose
+ * serde can report `Incompatible`, and those keys are registered, so they need no special case.
+ */
+object CometConfValidator extends Logging {
+
+  private val cometKeyPrefix = s"${CometConf.COMET_PREFIX}."
+
+  private val passThroughPrefixes = Seq(s"${CometConf.COMET_PREFIX}.datafusion.")
+
+  private val maxSuggestionDistance = 3
+
+  /** Keys already checked, so each one is checked, and warned about, at most once per JVM. */
+  private[comet] val checkedKeys = ConcurrentHashMap.newKeySet[String]()
+
+  /** Set when building the known keys fails, so a warning-only check never breaks a query. */
+  @volatile private var disabled = false
+
+  private lazy val knownKeys: Set[String] = {
+    // Loading the providers constructs the contrib config objects, which registers their entries
+    val entries = CometConfigProvider.providers.flatMap(_.configs) ++ CometConf.allConfs.toList
+    val exprNames = (QueryPlanSerde.exprSerdeMap.keys ++ QueryPlanSerde.aggrSerdeMap.keys)
+      .map(_.getSimpleName)
+    val exprKeys = exprNames.flatMap { name =>
+      Seq(CometConf.getExprEnabledConfigKey(name), CometConf.getExprAllowIncompatConfigKey(name))
+    }
+    (entries.flatMap(entry => entry.key +: entry.alternatives) ++ exprKeys ++
+      Seq(CometDriverPlugin.COMET_VERSION_CONFIG)).toSet
+  }
+
+  /** Known keys in sorted order with their lowercase form, so ties go to the first key. */
+  private lazy val suggestionCandidates: Seq[(String, UTF8String)] =
+    knownKeys.toSeq.sorted.map(key => key -> UTF8String.fromString(key.toLowerCase(Locale.ROOT)))
+
+  /** Unrecognized keys in sorted order, each with the closest known key if one is close. */
+  private[comet] def findUnrecognized(keys: Iterable[String]): Seq[(String, Option[String])] =
+    keys.filter(isUnrecognized).toSeq.distinct.sorted.map(key => key -> suggest(key))
+
+  /** Logs a warning for each unrecognized key that has not been checked yet. */
+  def warnUnrecognized(keys: Iterable[String]): Unit = {
+    if (disabled) return
+    val unchecked = keys.filter(key => key.startsWith(cometKeyPrefix) && checkedKeys.add(key))
+    if (unchecked.nonEmpty) {
+      try {
+        findUnrecognized(unchecked).foreach { case (key, suggestion) =>
+          val hint =
+            suggestion.map(s => s"Did you mean '$s'?").getOrElse("Check the key for typos.")
+          logWarning(s"Comet configuration '$key' is not recognized. $hint")
+        }
+      } catch {
+        case e @ (NonFatal(_) | _: LinkageError) =>
+          disabled = true
+          logDebug("Could not check Comet configuration keys", e)
+      }
+    }
+  }
+
+  private def isUnrecognized(key: String): Boolean =
+    key.startsWith(cometKeyPrefix) && !passThroughPrefixes.exists(key.startsWith) &&
+      !knownKeys.contains(key)
+
+  private def suggest(key: String): Option[String] = {
+    val lowerKey = UTF8String.fromString(key.toLowerCase(Locale.ROOT))
+    val (closest, distance) = suggestionCandidates
+      .map { case (known, lowerKnown) => known -> lowerKey.levenshteinDistance(lowerKnown) }
+      .minBy(_._2)
+    if (distance <= maxSuggestionDistance) Some(closest) else None
+  }
+}

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -57,7 +57,7 @@ import org.apache.spark.sql.execution.window.{WindowExec, WindowExecBase}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
-import org.apache.comet.{CometConf, CometExplainInfo, ConfigEntry, ExtendedExplainInfo}
+import org.apache.comet.{CometConf, CometConfValidator, CometExplainInfo, ConfigEntry, ExtendedExplainInfo}
 import org.apache.comet.CometConf.{COMET_SPARK_TO_ARROW_ENABLED, COMET_SPARK_TO_ARROW_SUPPORTED_OPERATOR_LIST}
 import org.apache.comet.CometSparkSessionExtensions._
 import org.apache.comet.rules.CometExecRule.allExecs
@@ -1099,6 +1099,10 @@ case class CometExecRule(session: SparkSession, queryStagePrep: Boolean = false)
   }
 
   private def _apply(plan: SparkPlan): SparkPlan = {
+    // The session conf covers SET, spark.conf.set and sessions without the driver plugin. This
+    // runs before the checks below, since a misspelled key can be why Comet is disabled.
+    CometConfValidator.warnUnrecognized(conf.getAllConfs.keys)
+
     // We shouldn't transform Spark query plan if Comet is not loaded.
     if (!isCometLoaded(conf)) {
       explainSparkReadsOfCometCache(plan, "Comet is disabled")

spark/src/main/scala/org/apache/spark/Plugins.scala

@@ -36,7 +36,7 @@ import org.apache.spark.sql.comet.execution.shuffle.{CometCelebornShuffleManager
 import org.apache.spark.sql.internal.StaticSQLConf
 import org.apache.spark.util.{Clock, SystemClock}
 
-import org.apache.comet.{COMET_VERSION, CometExecIterator, CometExecutorMemoryUsage, CometSparkSessionExtensions, NativeBase}
+import org.apache.comet.{COMET_VERSION, CometConfValidator, CometExecIterator, CometExecutorMemoryUsage, CometSparkSessionExtensions, NativeBase}
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometConf.{COMET_ICEBERG_WRITE_REPORT_DIR, COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
 import org.apache.comet.CometExecIterator.MemoryUsageSummary
@@ -104,6 +104,9 @@ class CometDriverPlugin private[spark] (clock: Clock) extends DriverPlugin with
     // disabled.
     sc.conf.set(CometDriverPlugin.COMET_VERSION_CONFIG, COMET_VERSION)
 
+    // Before the off-heap check, since a misspelled key can be why Comet is disabled
+    CometConfValidator.warnUnrecognized(sc.getConf.getAll.map(_._1).toSeq)
+
     if (!CometSparkSessionExtensions.isOffHeapEnabled(sc.getConf) &&
       !sc.getConf.getBoolean(COMET_ONHEAP_ENABLED.key, false)) {
       logWarning("Comet plugin is disabled because Spark is not running in off-heap mode.")

spark/src/test/resources/sql-tests/expressions/window/lag_lead.sql

@@ -15,8 +15,6 @@
 -- specific language governing permissions and limitations
 -- under the License.
 
--- Config: spark.comet.operator.WindowExec.allowIncompatible=true
-
 -- ============================================================
 -- Setup: shared tables
 -- ============================================================

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

@@ -296,4 +296,66 @@ class CometConfSuite extends AnyFunSuite {
 
     assert(CometConf.COMET_EXPLAIN_FALLBACK_LOG_ENABLED.get(conf))
   }
+
+  test("unrecognized keys are reported with the closest known key as a suggestion") {
+    assert(
+      CometConfValidator.findUnrecognized(
+        Seq(
+          "spark.comet.exec.enabeld",
+          "spark.comet.Exec.enabled",
+          "spark.comet.expression.Castt.enabled",
+          "spark.comet.foo.bar")) == Seq(
+        "spark.comet.Exec.enabled" -> Some("spark.comet.exec.enabled"),
+        "spark.comet.exec.enabeld" -> Some("spark.comet.exec.enabled"),
+        "spark.comet.expression.Castt.enabled" -> Some("spark.comet.expression.Cast.enabled"),
+        "spark.comet.foo.bar" -> None))
+  }
+
+  test("registered keys, alternatives and Comet-set keys are recognized") {
+    assert(
+      CometConfValidator
+        .findUnrecognized(
+          Seq(
+            CometConf.COMET_EXEC_ENABLED.key,
+            "spark.comet.exec.shuffle.enabled",
+            "spark.comet.version"))
+        .isEmpty)
+  }
+
+  test("expression keys are recognized only for expressions Comet has a serde for") {
+    assert(
+      CometConfValidator
+        .findUnrecognized(Seq(
+          "spark.comet.expression.Cast.enabled",
+          "spark.comet.expression.Cast.allowIncompatible",
+          "spark.comet.expression.Sum.enabled",
+          "spark.comet.expression.Sum.allowIncompatible"))
+        .isEmpty)
+    assert(
+      CometConfValidator
+        .findUnrecognized(Seq("spark.comet.expression.NoSuchExpression.enabled"))
+        .map(_._1) == Seq("spark.comet.expression.NoSuchExpression.enabled"))
+  }
+
+  test("operator keys are recognized only when registered") {
+    assert(
+      CometConfValidator
+        .findUnrecognized(Seq("spark.comet.operator.DataWritingCommandExec.allowIncompatible"))
+        .isEmpty)
+    assert(
+      CometConfValidator
+        .findUnrecognized(Seq("spark.comet.operator.WindowExec.allowIncompatible"))
+        .map(_._1) == Seq("spark.comet.operator.WindowExec.allowIncompatible"))
+  }
+
+  test("DataFusion pass-through keys and keys outside spark.comet are not checked") {
+    assert(
+      CometConfValidator
+        .findUnrecognized(
+          Seq(
+            "spark.comet.datafusion.execution.batch_size",
+            "spark.sql.shuffle.partitions",
+            "spark.cometx.foo"))
+        .isEmpty)
+  }
 }

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

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import org.apache.logging.log4j.Level
+import org.apache.spark.sql.CometTestBase
+
+class CometConfValidatorSuite extends CometTestBase {
+
+  private def warningsDuring(keys: String*)(f: => Unit): Seq[String] = {
+    keys.foreach(CometConfValidator.checkedKeys.remove)
+    // Logging derives the logger name by stripping the object's trailing '$'
+    val logger = CometConfValidator.getClass.getName.stripSuffix("$")
+    val appender = new LogAppender("unrecognized Comet configuration warning")
+    withLogAppender(appender, Seq(logger), Some(Level.WARN))(f)
+    // Keys that the test base sets for every test may warn too, so keep only these keys
+    appender.loggingEvents
+      .map(_.getMessage.getFormattedMessage)
+      .filter(message => keys.exists(key => message.contains(s"'$key'")))
+      .toSeq
+  }
+
+  test("warns once per key set on the session, with a suggestion") {
+    val key = "spark.comet.exec.enabeld"
+    val warnings = warningsDuring(key) {
+      withSQLConf(key -> "true") {
+        spark.range(1).count()
+        spark.range(2).count()
+      }
+    }
+    assert(
+      warnings == Seq(s"Comet configuration '$key' is not recognized. " +
+        "Did you mean 'spark.comet.exec.enabled'?"))
+  }
+
+  test("warns when Comet is disabled, since a misspelled key can be why") {
+    val key = "spark.comet.foo.bar"
+    val warnings = warningsDuring(key) {
+      withSQLConf(CometConf.COMET_ENABLED.key -> "false", key -> "true") {
+        spark.range(1).count()
+      }
+    }
+    assert(
+      warnings == Seq(s"Comet configuration '$key' is not recognized. Check the key for typos."))
+  }
+
+  test("does not warn for recognized keys") {
+    val keys = Seq("spark.comet.expression.Cast.enabled", CometConf.COMET_EXEC_ENABLED.key)
+    val warnings = warningsDuring(keys: _*) {
+      withSQLConf(keys.map(_ -> "true"): _*) {
+        spark.range(1).count()
+      }
+    }
+    assert(warnings.isEmpty)
+  }
+}

spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala

@@ -2032,7 +2032,6 @@ class CometAggregateSuite extends CometTestBase with AdaptiveSparkPlanHelper {
 
       withSQLConf(
         CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
-        "spark.comet.exec.shuffle.fallbackToColumnar" -> "false",
         "spark.comet.enabled" -> "true") {
         spark.read.parquet(dir.getAbsolutePath).createOrReplaceTempView("t2")
         checkSparkAnswerAndOperator("SELECT i, sum(v1), count(distinct v) FROM t2 group by i")

spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala

@@ -48,7 +48,6 @@ class CometWindowExecSuite extends CometTestBase {
       withSQLConf(
         CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
         CometConf.COMET_EXEC_WINDOW_ENABLED.key -> "true",
-        "spark.comet.operator.WindowExec.allowIncompatible" -> "true",
         "spark.comet.explain.fallback.enabled" -> "true",
         "spark.comet.explain.fallback.log.enabled" -> "true",
         "spark.comet.exec.localTableScan.enabled" -> "true") {

spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala

@@ -55,8 +55,7 @@ object CometSlidingSumBenchmark extends CometBenchmarkBase {
       SQLConf.SHUFFLE_PARTITIONS.key -> "1",
       CometConf.COMET_ENABLED.key -> "true",
       CometConf.COMET_EXEC_ENABLED.key -> "true",
-      CometConf.COMET_SHUFFLE_ENABLED.key -> "true",
-      "spark.comet.operator.WindowExec.allowIncompatible" -> "true") {
+      CometConf.COMET_SHUFFLE_ENABLED.key -> "true") {
       for ((shape, value) <- Seq(
           "cancelling" -> "id % 17 - 8",
           "positive" -> "id % 17",