This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new 957886db83 feat: add withAlternative alias mechanism for CometConf
(#4979)
957886db83 is described below
commit 957886db83c631027fffd11224a50de233f7bcf6
Author: Andy Grove <[email protected]>
AuthorDate: Mon Jul 20 15:39:25 2026 -0600
feat: add withAlternative alias mechanism for CometConf (#4979)
* feat: add withAlternative alias mechanism for CometConf, rename
replaceSortMergeJoin
Introduces `.withAlternative(oldKey, ...)` on `ConfigBuilder`. Reads check
the primary
key first, then any registered alternatives in order; reading a value from
an alternative
logs a one-time deprecation warning per JVM naming the current key.
Threaded through
`TypedConfigBuilder` into `ConfigEntryWithDefault` and
`OptionalConfigEntry` so the same
mechanism covers configs with defaults, env-var overrides, and optional
configs.
Pilots the mechanism by renaming `spark.comet.exec.replaceSortMergeJoin` to
`spark.comet.exec.forceShuffleHashJoin.enabled` (Scala symbol
`COMET_FORCE_SHJ`), keeping
the old key working as an alias.
Adds `CometConfSuite` with tests for primary-wins, alternative fallback,
default,
multiple alternatives, OptionalConfigEntry, and type conversion through the
alias path.
Adds a Configuration Conventions contributor-guide page that documents the
key/symbol
naming rules and the rename workflow.
Broader rename backlog tracked in #4978, per the 1.0.0 tracker #4082.
* fixup: forceShuffleHashJoin → forceShuffledHashJoin, drop .enabled suffix
Rename correction: use the past-participle form matching Spark's
`ShuffledHashJoin`
class. Also drops the `.enabled` suffix — the `force` verb already encodes
the
boolean action, so `.enabled` is redundant for this key.
Softens the Configuration Conventions doc's boolean-suffix rule accordingly:
descriptor-form flags (debug.enabled, metrics.enabled) still end in
`.enabled`;
action-verb flags (force..., allow...) can omit it.
* ci: register CometConfSuite in linux and macos PR workflows
---
.github/workflows/pr_build_linux.yml | 1 +
.github/workflows/pr_build_macos.yml | 1 +
.../contributor-guide/benchmark-results/tpc-h.md | 2 +-
.../source/contributor-guide/benchmarking_macos.md | 2 +-
.../source/contributor-guide/config_conventions.md | 115 +++++++++++++++++++
docs/source/contributor-guide/index.md | 1 +
docs/source/user-guide/latest/tuning.md | 2 +-
.../main/scala/org/apache/comet/CometConf.scala | 84 ++++++++++++--
.../org/apache/comet/rules/CometExecRule.scala | 2 +-
.../scala/org/apache/comet/CometConfSuite.scala | 124 +++++++++++++++++++++
10 files changed, 318 insertions(+), 16 deletions(-)
diff --git a/.github/workflows/pr_build_linux.yml
b/.github/workflows/pr_build_linux.yml
index 0cfe98d9e7..6aaac53204 100644
--- a/.github/workflows/pr_build_linux.yml
+++ b/.github/workflows/pr_build_linux.yml
@@ -340,6 +340,7 @@ jobs:
org.apache.comet.exec.CometJoinSuite
org.apache.spark.sql.comet.CometMapInBatchSuite
org.apache.comet.CometNativeSuite
+ org.apache.comet.CometConfSuite
org.apache.comet.CometSetOpWithGroupBySuite
org.apache.comet.CometSparkSessionExtensionsSuite
org.apache.spark.CometPluginsSuite
diff --git a/.github/workflows/pr_build_macos.yml
b/.github/workflows/pr_build_macos.yml
index e409da4b24..36ed5f9405 100644
--- a/.github/workflows/pr_build_macos.yml
+++ b/.github/workflows/pr_build_macos.yml
@@ -156,6 +156,7 @@ jobs:
org.apache.comet.exec.CometJoinSuite
org.apache.spark.sql.comet.CometMapInBatchSuite
org.apache.comet.CometNativeSuite
+ org.apache.comet.CometConfSuite
org.apache.comet.CometSetOpWithGroupBySuite
org.apache.comet.CometSparkSessionExtensionsSuite
org.apache.spark.CometPluginsSuite
diff --git a/docs/source/contributor-guide/benchmark-results/tpc-h.md
b/docs/source/contributor-guide/benchmark-results/tpc-h.md
index c489f8c999..eab15fb84a 100644
--- a/docs/source/contributor-guide/benchmark-results/tpc-h.md
+++ b/docs/source/contributor-guide/benchmark-results/tpc-h.md
@@ -74,6 +74,6 @@ spark.memory.offHeap.size=32G
### Comet (Tuned)
```properties
-spark.comet.exec.replaceSortMergeJoin=true
+spark.comet.exec.forceShuffledHashJoin=true
spark.comet.memoryPool.fraction=0.8
```
diff --git a/docs/source/contributor-guide/benchmarking_macos.md
b/docs/source/contributor-guide/benchmarking_macos.md
index 5e98acaae9..70135ed210 100644
--- a/docs/source/contributor-guide/benchmarking_macos.md
+++ b/docs/source/contributor-guide/benchmarking_macos.md
@@ -154,7 +154,7 @@ $SPARK_HOME/bin/spark-submit \
--conf spark.comet.enabled=true \
--conf spark.comet.exec.shuffle.enableFastEncoding=true \
--conf spark.comet.exec.shuffle.fallbackToColumnar=true \
- --conf spark.comet.exec.replaceSortMergeJoin=true \
+ --conf spark.comet.exec.forceShuffledHashJoin=true \
--conf spark.comet.expression.allowIncompatible=true \
$DF_BENCH/runners/datafusion-comet/tpcbench.py \
--benchmark tpch \
diff --git a/docs/source/contributor-guide/config_conventions.md
b/docs/source/contributor-guide/config_conventions.md
new file mode 100644
index 0000000000..0e01c958c1
--- /dev/null
+++ b/docs/source/contributor-guide/config_conventions.md
@@ -0,0 +1,115 @@
+<!--
+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.
+-->
+
+# Configuration Conventions
+
+Comet configuration keys live under `spark.comet.*` and are declared in
+`spark/src/main/scala/org/apache/comet/CometConf.scala`. This page describes
the naming
+conventions those keys follow so that new configs stay consistent with the
surface users
+already know, and documents the process for renaming an existing key without
breaking
+deployments.
+
+## Key Naming
+
+A Comet config key is a dotted path of segments. Each segment describes a
scope, from
+broadest (the prefix) to narrowest (the specific setting).
+
+- **Prefix.** All keys start with `spark.comet.` — never bare `comet.` and
never a nested
+ prefix like `spark.sql.comet.`.
+- **Category segment.** The first segment after the prefix names a broad area:
+ `exec` (execution and expressions), `scan` (source readers), `parquet`
(Parquet-specific
+ settings), `shuffle` (shuffle behavior), `columnar` (columnar shuffle),
`explain` (plan
+ explain/logging), `metrics`, `tracing`, `testing`, `convert` (Spark → Comet
source
+ conversion), or `expression` / `operator` (per-expression / per-operator
flags).
+- **Segment casing.** Multi-word segments use `camelCase`, not dot- or
kebab-separated.
+ Prefer `spark.comet.foo.maxThreadNum` over `spark.comet.foo.max.thread.num`
or
+ `spark.comet.foo.max-thread-num`.
+- **Boolean suffix.** Descriptor-form boolean flags that gate a subsystem or
feature end
+ in `.enabled` — for example `spark.comet.debug.enabled`,
`spark.comet.metrics.enabled`,
+ `spark.comet.parquet.rowFilterPushdown.enabled`. Action-form flags whose
name is itself
+ a verb (`force...`, `allow...`) can omit `.enabled` because the verb already
encodes
+ the action — for example `spark.comet.exec.forceShuffledHashJoin`.
+- **Acronyms.** Treat acronyms consistently as `UDF`, `SHJ`, `SMJ`, `IO` —
either all-caps
+ or camelCase, but do not mix within one key (`pyarrowUdf` and `scalaUDF` in
the same
+ cluster is a red flag).
+
+## Symbol Naming (Scala)
+
+Every config declares a Scala `val` in `CometConf.scala`. The symbol name is
the
+`UPPER_SNAKE_CASE` form of the key's meaningful segments, prefixed with
`COMET_`.
+
+Examples:
+
+| Key | Symbol
|
+| ----------------------------------------------- |
--------------------------- |
+| `spark.comet.enabled` | `COMET_ENABLED`
|
+| `spark.comet.exec.forceShuffledHashJoin` | `COMET_FORCE_SHJ`
|
+| `spark.comet.parquet.rowFilterPushdown.enabled` |
`COMET_ROW_FILTER_PUSHDOWN` |
+
+The symbol name is what appears in code; the key is what appears in user
configuration.
+The two do not have to match segment-for-segment — brevity in the symbol is
fine as long
+as the key remains descriptive.
+
+## Categories
+
+Every `ConfigEntry` must call `.category(...)`. The category is used to route
the key into
+the right table in the user guide's `configs.md`. Available categories are
declared as
+`CATEGORY_*` constants at the top of `CometConf.scala`. If a new config does
not fit an
+existing category, discuss adding a new one before landing the config.
+
+## Renaming an Existing Config
+
+Configs under `spark.comet.*` are stable across minor releases: users may have
set them in
+production `spark-defaults.conf` files, Spark job submissions, or notebooks.
Renaming a key
+must not silently break those deployments.
+
+Use the `withAlternative` builder on `ConfigBuilder` to keep the old key
working as a
+deprecated alias:
+
+```scala
+val COMET_FORCE_SHJ: ConfigEntry[Boolean] =
+ conf(s"$COMET_EXEC_CONFIG_PREFIX.forceShuffledHashJoin")
+ .withAlternative(s"$COMET_EXEC_CONFIG_PREFIX.replaceSortMergeJoin")
+ .category(CATEGORY_EXEC)
+ .doc("...")
+ .booleanConf
+ .createWithDefault(false)
+```
+
+Reading a value from the alternative logs a one-time deprecation warning per
JVM per
+alternative key, pointing users at the current key. The primary key always
wins if both
+are set.
+
+`withAlternative` accepts multiple alternatives (checked in order), for keys
that have
+been renamed more than once.
+
+The rename checklist for a single config:
+
+1. Update the key string on the `conf(...)` call and add
`.withAlternative(oldKey)`.
+2. Rename the Scala `val` (for example `COMET_REPLACE_SMJ` →
`COMET_FORCE_SHJ`).
+3. Update every callsite of the Scala symbol — grep the whole repo.
+4. Update every documented reference to the old key string — grep
`docs/source/` for the
+ old key and update to the new key. The auto-generated `configs.md` table
will refresh
+ on the next release-docs regeneration and does not need manual editing.
+5. Add or update a test in `CometConfSuite` if the rename covers a new type of
alias
+ pattern (single-hop, multiple alternatives, etc.).
+
+Removing a deprecated alias is a follow-up step that belongs to a later major
release —
+typically the next Comet major after the rename first ships. See the
+[versioning policy](../about/versioning_policy.md) for the timing rules.
diff --git a/docs/source/contributor-guide/index.md
b/docs/source/contributor-guide/index.md
index f293961bda..397239dbe4 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -91,6 +91,7 @@ Tracing <tracing>
Expression Audits <expression-audits/index>
Supported Spark Configurations <spark_configs_support>
+Configuration Conventions <config_conventions>
```
```{toctree}
diff --git a/docs/source/user-guide/latest/tuning.md
b/docs/source/user-guide/latest/tuning.md
index 5fa9c5f344..6f0e9a684b 100644
--- a/docs/source/user-guide/latest/tuning.md
+++ b/docs/source/user-guide/latest/tuning.md
@@ -119,7 +119,7 @@ to configure Comet to convert `SortMergeJoin` to
`ShuffledHashJoin`. Comet does
`ShuffledHashJoin` so this could result in OOM. Also, `SortMergeJoin` may
still be faster in some cases. It is best
to test with both for your specific workloads.
-To configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`, set
`spark.comet.exec.replaceSortMergeJoin=true`.
+To configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`, set
`spark.comet.exec.forceShuffledHashJoin=true`.
## Shuffle
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala
b/spark/src/main/scala/org/apache/comet/CometConf.scala
index e49cf0db1d..b1fbeea5cb 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -326,8 +326,9 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
- val COMET_REPLACE_SMJ: ConfigEntry[Boolean] =
- conf(s"$COMET_EXEC_CONFIG_PREFIX.replaceSortMergeJoin")
+ val COMET_FORCE_SHJ: ConfigEntry[Boolean] =
+ conf(s"$COMET_EXEC_CONFIG_PREFIX.forceShuffledHashJoin")
+ .withAlternative(s"$COMET_EXEC_CONFIG_PREFIX.replaceSortMergeJoin")
.category(CATEGORY_EXEC)
.doc("Experimental feature to force Spark to replace SortMergeJoin with
ShuffledHashJoin " +
s"for improved performance. This feature is not stable yet.
$TUNING_GUIDE.")
@@ -1026,7 +1027,8 @@ private class TypedConfigBuilder[T](
parent._doc,
parent._category,
parent._public,
- parent._version)
+ parent._version,
+ parent._alternatives)
CometConf.register(conf)
conf
}
@@ -1042,7 +1044,9 @@ private class TypedConfigBuilder[T](
parent._doc,
parent._category,
parent._public,
- parent._version)
+ parent._version,
+ None,
+ parent._alternatives)
CometConf.register(conf)
conf
}
@@ -1072,7 +1076,8 @@ private class TypedConfigBuilder[T](
parent._category,
parent._public,
parent._version,
- Some(envVar))
+ Some(envVar),
+ parent._alternatives)
CometConf.register(conf)
conf
}
@@ -1085,7 +1090,8 @@ abstract class ConfigEntry[T](
val doc: String,
val category: String,
val isPublic: Boolean,
- val version: String) {
+ val version: String,
+ val alternatives: Seq[String] = Nil) {
/**
* Retrieves the config value from the given [[SQLConf]].
@@ -1123,8 +1129,17 @@ private[comet] class ConfigEntryWithDefault[T](
category: String,
isPublic: Boolean,
version: String,
- _envVar: Option[String] = None)
- extends ConfigEntry(key, valueConverter, stringConverter, doc, category,
isPublic, version) {
+ _envVar: Option[String] = None,
+ _alternatives: Seq[String] = Nil)
+ extends ConfigEntry(
+ key,
+ valueConverter,
+ stringConverter,
+ doc,
+ category,
+ isPublic,
+ version,
+ _alternatives) {
override def defaultValue: Option[T] = Some(_defaultValue)
override def defaultValueString: String = stringConverter(_defaultValue)
@@ -1132,7 +1147,7 @@ private[comet] class ConfigEntryWithDefault[T](
override def envVar: Option[String] = _envVar
def get(conf: SQLConf): T = {
- val tmp = conf.getConfString(key, null)
+ val tmp = CometConfDeprecations.readWithAlternatives(conf, key,
alternatives)
if (tmp == null) {
_defaultValue
} else {
@@ -1148,7 +1163,8 @@ private[comet] class OptionalConfigEntry[T](
doc: String,
category: String,
isPublic: Boolean,
- version: String)
+ version: String,
+ _alternatives: Seq[String] = Nil)
extends ConfigEntry[Option[T]](
key,
s => Some(rawValueConverter(s)),
@@ -1156,12 +1172,14 @@ private[comet] class OptionalConfigEntry[T](
doc,
category,
isPublic,
- version) {
+ version,
+ _alternatives) {
override def defaultValueString: String = ConfigEntry.UNDEFINED
override def get(conf: SQLConf): Option[T] = {
- Option(conf.getConfString(key, null)).map(rawValueConverter)
+ Option(CometConfDeprecations.readWithAlternatives(conf, key, alternatives))
+ .map(rawValueConverter)
}
}
@@ -1173,12 +1191,23 @@ private[comet] case class ConfigBuilder(key: String) {
var _doc = ""
var _version = ""
var _category = ""
+ var _alternatives: Seq[String] = Nil
def internal(): ConfigBuilder = {
_public = false
this
}
+ /**
+ * Registers deprecated config keys that Comet will read as fall-backs when
the primary `key` is
+ * unset. Reading a value from an alternative logs a one-time deprecation
warning pointing users
+ * at the current `key`. Alternatives are checked in the order provided.
+ */
+ def withAlternative(alt: String, more: String*): ConfigBuilder = {
+ _alternatives = alt +: more
+ this
+ }
+
def doc(s: String): ConfigBuilder = {
_doc = s
this
@@ -1226,3 +1255,34 @@ private[comet] case class ConfigBuilder(key: String) {
private object ConfigEntry {
val UNDEFINED = "<undefined>"
}
+
+private object CometConfDeprecations extends org.apache.spark.internal.Logging
{
+
+ private val warned =
java.util.concurrent.ConcurrentHashMap.newKeySet[String]()
+
+ /**
+ * Reads the config value for `key`, falling back to each key in
`alternatives` (in order) when
+ * the primary key is unset. When the value comes from an alternative, logs
a deprecation
+ * warning once per JVM per alternative key.
+ *
+ * @return
+ * the resolved string value, or `null` if neither the primary key nor any
alternative is set.
+ */
+ def readWithAlternatives(conf: SQLConf, key: String, alternatives:
Seq[String]): String = {
+ val primary = conf.getConfString(key, null)
+ if (primary != null) return primary
+ if (alternatives.isEmpty) return null
+ val it = alternatives.iterator
+ while (it.hasNext) {
+ val alt = it.next()
+ val v = conf.getConfString(alt, null)
+ if (v != null) {
+ if (warned.add(alt)) {
+ logWarning(s"Comet configuration '$alt' is deprecated; use '$key'
instead.")
+ }
+ return v
+ }
+ }
+ null
+ }
+}
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 4c8e68cc0d..5d4252d6b2 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala
@@ -573,7 +573,7 @@ case class CometExecRule(session: SparkSession)
} else {
val normalizedPlan = normalizePlan(plan)
- val planWithJoinRewritten = if (CometConf.COMET_REPLACE_SMJ.get()) {
+ val planWithJoinRewritten = if (CometConf.COMET_FORCE_SHJ.get()) {
normalizedPlan.transformUp { case p =>
RewriteJoin.rewrite(p)
}
diff --git a/spark/src/test/scala/org/apache/comet/CometConfSuite.scala
b/spark/src/test/scala/org/apache/comet/CometConfSuite.scala
new file mode 100644
index 0000000000..4757dca83f
--- /dev/null
+++ b/spark/src/test/scala/org/apache/comet/CometConfSuite.scala
@@ -0,0 +1,124 @@
+/*
+ * 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.scalatest.funsuite.AnyFunSuite
+
+import org.apache.spark.sql.internal.SQLConf
+
+class CometConfSuite extends AnyFunSuite {
+
+ test("primary key wins over alternative when both are set") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.primaryWins")
+ .withAlternative("spark.comet.testing.alias.primaryWins.old")
+ .category("testing")
+ .booleanConf
+ .createWithDefault(false)
+
+ val conf = new SQLConf
+ conf.setConfString(entry.key, "true")
+ conf.setConfString(entry.alternatives.head, "false")
+
+ assert(entry.get(conf))
+ }
+
+ test("alternative is read when primary key is unset, with expected value") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.readsAlternative")
+ .withAlternative("spark.comet.testing.alias.readsAlternative.old")
+ .category("testing")
+ .intConf
+ .createWithDefault(0)
+
+ val conf = new SQLConf
+ conf.setConfString(entry.alternatives.head, "42")
+
+ assert(entry.get(conf) == 42)
+ }
+
+ test("default is returned when neither primary nor alternative is set") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.defaultOnly")
+ .withAlternative("spark.comet.testing.alias.defaultOnly.old")
+ .category("testing")
+ .booleanConf
+ .createWithDefault(true)
+
+ val conf = new SQLConf
+ assert(entry.get(conf))
+ }
+
+ test("multiple alternatives are checked in the order provided") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.multi")
+ .withAlternative(
+ "spark.comet.testing.alias.multi.older",
+ "spark.comet.testing.alias.multi.oldest")
+ .category("testing")
+ .intConf
+ .createWithDefault(0)
+
+ val conf = new SQLConf
+ conf.setConfString(entry.alternatives.head, "1")
+ conf.setConfString(entry.alternatives(1), "2")
+
+ // First alternative wins, not the second.
+ assert(entry.get(conf) == 1)
+ }
+
+ test("OptionalConfigEntry reads through an alternative") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.optional")
+ .withAlternative("spark.comet.testing.alias.optional.old")
+ .category("testing")
+ .stringConf
+ .createOptional
+
+ val conf = new SQLConf
+ assert(entry.get(conf).isEmpty)
+
+ conf.setConfString(entry.alternatives.head, "hello")
+ assert(entry.get(conf).contains("hello"))
+ }
+
+ test("value from an alternative goes through the type converter") {
+ val entry = CometConf
+ .conf("spark.comet.testing.alias.typed")
+ .withAlternative("spark.comet.testing.alias.typed.old")
+ .category("testing")
+ .booleanConf
+ .createWithDefault(false)
+
+ val conf = new SQLConf
+ conf.setConfString(entry.alternatives.head, "TRUE")
+
+ // The boolean converter accepts case-insensitive "TRUE"/"FALSE"; if the
alternative
+ // were returned raw, this assertion would fail.
+ assert(entry.get(conf))
+ }
+
+ test("COMET_FORCE_SHJ reads the deprecated replaceSortMergeJoin key as an
alias") {
+ val conf = new SQLConf
+
conf.setConfString(s"${CometConf.COMET_EXEC_CONFIG_PREFIX}.replaceSortMergeJoin",
"true")
+
+ assert(CometConf.COMET_FORCE_SHJ.get(conf))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
