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 936794d7f6 refactor: rename pyarrowUdf config to pyarrowUDF and fix
stale config docs (#5197)
936794d7f6 is described below
commit 936794d7f6884e4a71665d1c4390b8877a43a7ee
Author: Andy Grove <[email protected]>
AuthorDate: Mon Aug 3 07:16:26 2026 -0600
refactor: rename pyarrowUdf config to pyarrowUDF and fix stale config docs
(#5197)
---
.../source/contributor-guide/config_conventions.md | 27 +++++++++++-----------
docs/source/user-guide/latest/pyarrow-udfs.md | 8 +++----
.../main/scala/org/apache/comet/CometConf.scala | 2 +-
.../rules/EliminateRedundantTransitions.scala | 2 +-
.../resources/pyspark/benchmark_pyarrow_udf.py | 4 ++--
.../src/test/resources/pyspark/test_pyarrow_udf.py | 14 +++++------
6 files changed, 29 insertions(+), 28 deletions(-)
diff --git a/docs/source/contributor-guide/config_conventions.md
b/docs/source/contributor-guide/config_conventions.md
index 90042a71c0..5bcad087ea 100644
--- a/docs/source/contributor-guide/config_conventions.md
+++ b/docs/source/contributor-guide/config_conventions.md
@@ -34,9 +34,9 @@ broadest (the prefix) to narrowest (the specific setting).
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).
+ settings), `shuffle` (shuffle behavior), `explain` (plan explain/logging),
`metrics`,
+ `tracing`, `debug`, `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`.
@@ -46,8 +46,8 @@ broadest (the prefix) to narrowest (the specific setting).
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).
+ or camelCase, but do not mix within one cluster (`pyarrowUdf` alongside
`scalaUDF` would
+ be a red flag; both spell it `UDF`).
## Symbol Naming (Scala)
@@ -56,11 +56,11 @@ Every config declares a Scala `val` in `CometConf.scala`.
The symbol name is the
Examples:
-| Key | Symbol
|
-| ----------------------------------------------- |
--------------------------- |
-| `spark.comet.enabled` | `COMET_ENABLED`
|
-| `spark.comet.exec.forceShuffledHashJoin` | `COMET_FORCE_SHJ`
|
-| `spark.comet.parquet.rowFilterPushdown.enabled` |
`COMET_ROW_FILTER_PUSHDOWN` |
+| Key | Symbol
|
+| ----------------------------------------------- |
------------------------------------------- |
+| `spark.comet.enabled` | `COMET_ENABLED`
|
+| `spark.comet.exec.forceShuffledHashJoin` | `COMET_FORCE_SHJ`
|
+| `spark.comet.parquet.rowFilterPushdown.enabled` |
`COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED` |
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
@@ -137,9 +137,10 @@ val COMET_LEGACY_FOO_BEHAVIOR: ConfigEntry[Boolean] =
```
Naming follows the usual conventions: `spark.comet.legacy.` prefix,
`camelCase` final segment, and
-a `COMET_LEGACY_*` Scala symbol. The doc string must state which release
changed the behavior and
-what the old behavior was, so that `configs.md` is self-explanatory without
cross-referencing the
-upgrade guide.
+a `COMET_LEGACY_*` Scala symbol. No legacy config exists yet, so the first one
to land must also add
+the `CATEGORY_LEGACY` constant to `CometConf.scala` and a corresponding
section to `configs.md`. The
+doc string must state which release changed the behavior and what the old
behavior was, so that
+`configs.md` is self-explanatory without cross-referencing the upgrade guide.
The checklist for a behavior change:
diff --git a/docs/source/user-guide/latest/pyarrow-udfs.md
b/docs/source/user-guide/latest/pyarrow-udfs.md
index 31cf932107..ff493de23c 100644
--- a/docs/source/user-guide/latest/pyarrow-udfs.md
+++ b/docs/source/user-guide/latest/pyarrow-udfs.md
@@ -77,14 +77,14 @@ CometMapInBatch <- Arrow batch in/out, Python
runner attached
The optimization is experimental and disabled by default. Enable it with:
```
-spark.comet.exec.pyarrowUdf.enabled=true
+spark.comet.exec.pyarrowUDF.enabled=true
```
The default is `false` while the feature stabilizes.
### Relationship to Spark's PySpark Arrow conversion conf
-`spark.comet.exec.pyarrowUdf.enabled` is **not** the same as PySpark's
+`spark.comet.exec.pyarrowUDF.enabled` is **not** the same as PySpark's
[`spark.sql.execution.arrow.pyspark.enabled`](https://spark.apache.org/docs/latest/api/python/tutorial/sql/arrow_pandas.html#enabling-for-conversion-to-from-pandas).
That conf controls whether Spark uses Arrow when materializing a DataFrame to
a Pandas DataFrame
(`toPandas()`) or constructing one from Pandas. The Comet conf controls a
planner rewrite for
@@ -110,7 +110,7 @@ spark = SparkSession.builder \
.config("spark.plugins", "org.apache.spark.CometPlugin") \
.config("spark.comet.enabled", "true") \
.config("spark.comet.exec.enabled", "true") \
- .config("spark.comet.exec.pyarrowUdf.enabled", "true") \
+ .config("spark.comet.exec.pyarrowUDF.enabled", "true") \
.config("spark.memory.offHeap.enabled", "true") \
.config("spark.memory.offHeap.size", "2g") \
.getOrCreate()
@@ -201,7 +201,7 @@ on the unoptimized path.
UDF that reads the Arrow field's time zone or localizes to wall-clock time
(for example a
`mapInPandas` UDF that strips the tz and treats the value as naive local
time): under a non-UTC
session time zone such a UDF can diverge from the unoptimized path. Set
- `spark.comet.exec.pyarrowUdf.enabled=false` for those UDFs.
+ `spark.comet.exec.pyarrowUDF.enabled=false` for those UDFs.
- `spark.sql.execution.arrow.useLargeVarTypes=true` is not supported. With
this conf enabled,
Spark widens `StringType` and `BinaryType` to Arrow's 8-byte-offset variants
in the
destination IPC root, while Comet's source vectors always use 4-byte
offsets. The buffer-copy
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala
b/spark/src/main/scala/org/apache/comet/CometConf.scala
index ae29715193..e9cf8ca010 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -256,7 +256,7 @@ object CometConf extends ShimCometConf {
.createWithDefault(true)
val COMET_PYARROW_UDF_ENABLED: ConfigEntry[Boolean] =
- conf("spark.comet.exec.pyarrowUdf.enabled")
+ conf("spark.comet.exec.pyarrowUDF.enabled")
.category(CATEGORY_EXEC)
.doc(
"Experimental: whether to enable optimized execution of PyArrow UDFs "
+
diff --git
a/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
b/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
index 4b4b8c13bd..ec277cfc7b 100644
---
a/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
+++
b/spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala
@@ -190,7 +190,7 @@ case class EliminateRedundantTransitions(session:
SparkSession)
/**
* Matches the plans that could run natively as `CometMapInBatchExec`,
independent of whether
- * the `spark.comet.exec.pyarrowUdf.enabled` feature flag is set. The
`transformUp` arm reads
+ * the `spark.comet.exec.pyarrowUDF.enabled` feature flag is set. The
`transformUp` arm reads
* that flag to decide between rewriting the operator and merely annotating
it with an opt-in
* hint. Single extractor so the matchers run once per visited plan. Returns
`(info,
* columnarChild)` where `columnarChild` is the Comet columnar producer that
diff --git a/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
b/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
index dd6d8b3717..19d9dac5d3 100644
--- a/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
+++ b/spark/src/test/resources/pyspark/benchmark_pyarrow_udf.py
@@ -23,7 +23,7 @@ Requires PySpark 4.0.1+ (Comet's columnar runner targets
Spark 4.0+ only;
3.5 and 3.4 are documented no-ops).
Times `df.mapInArrow(passthrough, schema).count()` and the equivalent
-`mapInPandas` query with `spark.comet.exec.pyarrowUdf.enabled` set
+`mapInPandas` query with `spark.comet.exec.pyarrowUDF.enabled` set
to false (vanilla Spark path) and true (Comet's optimized path). Both
modes run the same Python worker, so the measured delta covers what the
optimization actually changes for users:
@@ -148,7 +148,7 @@ def _temp_parquet(spark: SparkSession, build_df, n: int):
def _time_run(spark: SparkSession, parquet_path: str, accelerate: bool, api:
str) -> float:
spark.conf.set(
- "spark.comet.exec.pyarrowUdf.enabled",
+ "spark.comet.exec.pyarrowUDF.enabled",
"true" if accelerate else "false",
)
df = spark.read.parquet(parquet_path)
diff --git a/spark/src/test/resources/pyspark/test_pyarrow_udf.py
b/spark/src/test/resources/pyspark/test_pyarrow_udf.py
index a7bdd598b3..84f405e8b2 100644
--- a/spark/src/test/resources/pyspark/test_pyarrow_udf.py
+++ b/spark/src/test/resources/pyspark/test_pyarrow_udf.py
@@ -20,9 +20,9 @@
Pytest-driven integration tests for Comet's PyArrow UDF acceleration.
Each test runs against two execution paths:
- - "accelerated": spark.comet.exec.pyarrowUdf.enabled=true
+ - "accelerated": spark.comet.exec.pyarrowUDF.enabled=true
(plan should contain CometMapInBatch and no ColumnarToRow)
- - "fallback": spark.comet.exec.pyarrowUdf.enabled=false
+ - "fallback": spark.comet.exec.pyarrowUDF.enabled=false
(plan should contain vanilla PythonMapInArrow / MapInArrow)
Usage:
@@ -82,7 +82,7 @@ def spark():
@pytest.fixture(params=[True, False], ids=["accelerated", "fallback"])
def accelerated(request, spark) -> bool:
spark.conf.set(
- "spark.comet.exec.pyarrowUdf.enabled",
+ "spark.comet.exec.pyarrowUDF.enabled",
"true" if request.param else "false",
)
return request.param
@@ -957,7 +957,7 @@ def
test_map_in_arrow_falls_back_when_use_large_var_types(spark, tmp_path):
setBytes per buffer and would corrupt the offset buffer in this
configuration.
EliminateRedundantTransitions must skip the rewrite in that case so
vanilla Spark
handles the operation. This test does not use the `accelerated` fixture:
it sets
- pyarrowUdf.enabled=true AND useLargeVarTypes=true and asserts the plan
still falls
+ pyarrowUDF.enabled=true AND useLargeVarTypes=true and asserts the plan
still falls
back to vanilla MapInArrow.
"""
schema_in = T.StructType(
@@ -974,9 +974,9 @@ def
test_map_in_arrow_falls_back_when_use_large_var_types(spark, tmp_path):
for batch in iterator:
yield batch
- prev_pyarrow = spark.conf.get("spark.comet.exec.pyarrowUdf.enabled",
"false")
+ prev_pyarrow = spark.conf.get("spark.comet.exec.pyarrowUDF.enabled",
"false")
prev_large = spark.conf.get("spark.sql.execution.arrow.useLargeVarTypes",
"false")
- spark.conf.set("spark.comet.exec.pyarrowUdf.enabled", "true")
+ spark.conf.set("spark.comet.exec.pyarrowUDF.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.useLargeVarTypes", "true")
try:
result_df = spark.read.parquet(src).mapInArrow(passthrough, schema_in)
@@ -991,7 +991,7 @@ def
test_map_in_arrow_falls_back_when_use_large_var_types(spark, tmp_path):
out = sorted((r["id"], r["name"]) for r in result_df.collect())
assert out == sorted(rows)
finally:
- spark.conf.set("spark.comet.exec.pyarrowUdf.enabled", prev_pyarrow)
+ spark.conf.set("spark.comet.exec.pyarrowUDF.enabled", prev_pyarrow)
spark.conf.set("spark.sql.execution.arrow.useLargeVarTypes",
prev_large)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
