From 99aff9593e4bc60f68636ae5ce62b44856743b5d Mon Sep 17 00:00:00 2001 From: tlfua <0050211@gmail.com> Date: Sat, 7 Nov 2026 18:04:06 -0800 Subject: [PATCH 1/2] feat: warn about unrecognized spark.comet configuration keys Log one warning per JVM for each explicitly set spark.comet.* key that Comet does not recognize, with a "Did you mean" hint when a known key is within Levenshtein distance 3. Keys are checked at driver plugin init and at the start of CometExecRule, before any early return, so a typo that disables Comet is still reported. spark.comet.datafusion.* keys are passed through to DataFusion and are not checked. Also removes test configs that set keys Comet no longer reads. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 94df0e1d-5dc9-4e71-93d5-d7b522aea46b --- .../adding_a_new_operator.md | 6 + docs/source/user-guide/latest/configs.md | 4 + .../org/apache/comet/CometConfValidator.scala | 107 ++++++++++++++++++ .../apache/comet/rules/CometExecRule.scala | 6 +- .../main/scala/org/apache/spark/Plugins.scala | 5 +- .../sql-tests/expressions/window/lag_lead.sql | 2 - .../org/apache/comet/CometConfSuite.scala | 62 ++++++++++ .../comet/CometConfValidatorSuite.scala | 73 ++++++++++++ .../comet/exec/CometAggregateSuite.scala | 1 - .../comet/exec/CometWindowExecSuite.scala | 1 - .../benchmark/CometSlidingSumBenchmark.scala | 3 +- 11 files changed, 262 insertions(+), 8 deletions(-) create mode 100644 spark/src/main/scala/org/apache/comet/CometConfValidator.scala create mode 100644 spark/src/test/scala/org/apache/comet/CometConfValidatorSuite.scala diff --git a/docs/source/contributor-guide/adding_a_new_operator.md b/docs/source/contributor-guide/adding_a_new_operator.md index b2beae1740b..9e20642fe86 100644 --- a/docs/source/contributor-guide/adding_a_new_operator.md +++ b/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..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`: diff --git a/docs/source/user-guide/latest/configs.md b/docs/source/user-guide/latest/configs.md index 0fb0ec56c49..601cba42e56 100644 --- a/docs/source/user-guide/latest/configs.md +++ b/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 diff --git a/spark/src/main/scala/org/apache/comet/CometConfValidator.scala b/spark/src/main/scala/org/apache/comet/CometConfValidator.scala new file mode 100644 index 00000000000..4e5d2d278f4 --- /dev/null +++ b/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 + } +} diff --git a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala index 70d108f2506..1af8022c925 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala +++ b/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") diff --git a/spark/src/main/scala/org/apache/spark/Plugins.scala b/spark/src/main/scala/org/apache/spark/Plugins.scala index 70f3ce6cedc..43777b14cc5 100644 --- a/spark/src/main/scala/org/apache/spark/Plugins.scala +++ b/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.") diff --git a/spark/src/test/resources/sql-tests/expressions/window/lag_lead.sql b/spark/src/test/resources/sql-tests/expressions/window/lag_lead.sql index 8f85174301e..bb45e7073da 100644 --- a/spark/src/test/resources/sql-tests/expressions/window/lag_lead.sql +++ b/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 -- ============================================================ diff --git a/spark/src/test/scala/org/apache/comet/CometConfSuite.scala b/spark/src/test/scala/org/apache/comet/CometConfSuite.scala index 5c269589310..86fb546f507 100644 --- a/spark/src/test/scala/org/apache/comet/CometConfSuite.scala +++ b/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) + } } diff --git a/spark/src/test/scala/org/apache/comet/CometConfValidatorSuite.scala b/spark/src/test/scala/org/apache/comet/CometConfValidatorSuite.scala new file mode 100644 index 00000000000..4daece010ff --- /dev/null +++ b/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) + } +} diff --git a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala index a5696212145..37a080869c7 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala +++ b/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") diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala index 5d09646d61e..3661b6e939b 100644 --- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala +++ b/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") { diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala index f533de2c7de..2f006fe225c 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometSlidingSumBenchmark.scala +++ b/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", From abef7ef578befcadd3d0dff58641c5c1de9f9cca Mon Sep 17 00:00:00 2001 From: tlfua <0050211@gmail.com> Date: Sat, 7 Nov 2026 18:08:32 -0800 Subject: [PATCH 2/2] ci: run CometConfValidatorSuite in PR builds Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 94df0e1d-5dc9-4e71-93d5-d7b522aea46b --- .github/workflows/pr_build_linux.yml | 1 + .github/workflows/pr_build_macos.yml | 1 + 2 files changed, 2 insertions(+) diff --git a/.github/workflows/pr_build_linux.yml b/.github/workflows/pr_build_linux.yml index ef593ff4ed5..939104a2b92 100644 --- a/.github/workflows/pr_build_linux.yml +++ b/.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 diff --git a/.github/workflows/pr_build_macos.yml b/.github/workflows/pr_build_macos.yml index 79cf33713ec..899d9810ea6 100644 --- a/.github/workflows/pr_build_macos.yml +++ b/.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