docs: expand tuning guide with performance and memory configs - #4908
Conversation
Shuffle-related configs were spread across four disjoint prefixes:
`spark.comet.exec.shuffle.*`, `spark.comet.columnar.shuffle.*`,
`spark.comet.native.shuffle.*`, and `spark.comet.shuffle.*`. This
consolidates all shuffle configs under `spark.comet.shuffle.*` with
`.columnar.` and `.native.` sub-namespaces for mode-specific settings.
Every renamed key is registered via `.withAlternative(...)`, so
existing user configurations continue to work (a deprecation warning
is logged when an old key is read). Per-mode leaves that were
previously misplaced move to their honest namespace:
- `spark.comet.exec.shuffle.writeBufferSize` (native-only) becomes
`spark.comet.shuffle.native.writeBufferSize`.
- `spark.comet.shuffle.preferDictionary.ratio` (JVM-only) becomes
`spark.comet.shuffle.columnar.preferDictionary.ratio`.
The `spark.comet.columnar.shuffle.*` keys also fix Category 2 in
apache#4978 (dots-in-segment): `spill.threshold`,
`memory.factor`, and `batch.size` become `spillThreshold`,
`memoryFactor`, and `batchSize`.
Part of apache#4978.
The JVM-side shuffle path in Comet is not the only columnar shuffle — native shuffle is also columnar (both use Arrow). The distinguishing axis is *where* the shuffle logic runs. Renaming the sub-namespace to `.jvm.` aligns it with the mode value users already set: `spark.comet.shuffle.mode = jvm`. Keys renamed: - spark.comet.shuffle.columnar.spillThreshold → spark.comet.shuffle.jvm.spillThreshold - spark.comet.shuffle.columnar.memoryFactor → spark.comet.shuffle.jvm.memoryFactor - spark.comet.shuffle.columnar.batchSize → spark.comet.shuffle.jvm.batchSize - spark.comet.shuffle.columnar.preferDictionary.ratio → spark.comet.shuffle.jvm.preferDictionary.ratio The pre-existing deprecated aliases (`spark.comet.columnar.shuffle.*`) are untouched — they were never `.columnar.`-scoped under the new prefix. Val names follow: `COMET_SHUFFLE_COLUMNAR_*` → `COMET_SHUFFLE_JVM_*`. The Scala/Java class names (`CometColumnarShuffle`, `CometColumnarShuffleSuite`, etc.) still use "columnar" and are left alone — that is a separate rename to consider once the config vocabulary settles.
Table column alignment tweaks after the `.columnar.` → `.jvm.` key rename. Content unchanged.
Resolves conflicts introduced by apache/main apache#4985 (remove untested async columnar shuffle) against this branch's shuffle-config renames: - Accept deletion of CometShuffleExternalSorterAsync/Sync and the merged-back CometShuffleExternalSorter; propagate this branch's jvm.* renames to the merged class (COMET_SHUFFLE_JVM_SPILL_THRESHOLD, COMET_SHUFFLE_JVM_PREFER_DICTIONARY_RATIO, COMET_SHUFFLE_COMPRESSION_*). - Drop async-config rows from jvm_shuffle.md and async-mode benchmark cases from CometShuffleBenchmark; drop the async-branch conditionals from CometColumnarShuffleSuite. - Fold the newly-introduced COMET_COLUMNAR_SHUFFLE_MAX_WRITERS_PER_EXECUTOR into this branch's naming: rename to COMET_SHUFFLE_JVM_MAX_WRITERS_PER_EXECUTOR (key spark.comet.shuffle.jvm.maxWritersPerExecutor), keeping both prior names as .withAlternative fallbacks. - Rename remaining spark.comet.exec.shuffle.mode occurrences in three make_time_shuffle*.sql test files to the new spark.comet.shuffle.mode.
# Conflicts: # spark/src/main/scala/org/apache/spark/sql/comet/execution/shuffle/CometNativeShuffleWriter.scala
Regenerate the Spark SQL test diffs to reference shuffle configs by their string keys rather than CometConf constants, so the config rename in this PR no longer breaks test compilation. - spark.comet.exec.shuffle.enabled -> spark.comet.shuffle.enabled - CometConf.COMET_EXEC_SHUFFLE_WITH_RANGE_PARTITIONING_ENABLED.key -> "spark.comet.shuffle.native.partitioning.range.enabled"
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks @andygrove!
|
|
||
| ## Parquet Reader Tuning | ||
|
|
||
| ### Parallel I/O |
There was a problem hiding this comment.
This subsection documents four configs that no longer exist on main. spark.comet.parquet.read.parallel.io.enabled, spark.comet.parquet.read.parallel.io.thread-pool.size, spark.comet.parquet.read.io.mergeRanges, and spark.comet.parquet.read.io.mergeRanges.delta were removed in b016555 (#4981), which merged on 2026-07-20, a week after this branch was cut. Grepping spark/, common/, and native/ on main finds no references to any of them outside generated target/ output. #4981 removed them specifically because they misled operators tuning benchmarks, so documenting them again would reverse that. Could you drop ### Parallel I/O entirely?
| ### Limiting Spill Disk Usage | ||
|
|
||
| Native operators that spill to disk (aggregate, sort, shuffle) are collectively bounded by | ||
| `spark.comet.maxTempDirectorySize` (default 100 GB per executor). If the limit is reached, further spills |
There was a problem hiding this comment.
Is "per executor" the right scope here? The DiskManagerBuilder carrying with_max_temp_directory_size is constructed in prepare_datafusion_session_context (native/core/src/execution/jni_api.rs:555), which createPlan calls once per native plan per task, and the comment just below notes the context is scoped to an executing Spark task. If that reading is right, each task gets its own 100 GB allowance and an executor running N concurrent tasks can reach N times that. The aggregate is what someone protecting a shared disk needs to reason about, so stating the per-task scope would make the last sentence of this paragraph actionable.
| .doc("The maximum amount of data (in bytes) stored inside the temporary directories.") | ||
| .category(CATEGORY_TUNING) | ||
| .doc( | ||
| "The maximum amount of data (in bytes) stored inside the temporary directories " + |
There was a problem hiding this comment.
Same scope question as tuning.md:120. The expanded doc string is clearer than what it replaces, and adding that the limit applies per task rather than per executor would make it complete, since this string is what shows up in the generated configs.md table.
| fail and the query errors out. Raise this on workloads with large sort/aggregate/shuffle spills, or lower | ||
| it to protect executors on shared disks. | ||
|
|
||
| ## Parquet Reader Tuning |
There was a problem hiding this comment.
Once this is rebased, main already has a ### Parquet Native Scans section (the row group split mismatch and issue #3817) currently nested under ## Shuffle at tuning.md:189. That leaves Parquet guidance in two places with one of them under the wrong parent. Would you move that section under this new ## Parquet Reader Tuning heading as part of the rebase?
| at higher CPU cost — useful when shuffle I/O or network bandwidth is the bottleneck. When ZSTD is selected, the | ||
| level is controlled by `spark.comet.exec.shuffle.compression.zstd.level` (default `1`). | ||
|
|
||
| ### Reducing Row/Columnar Conversion Overhead |
There was a problem hiding this comment.
This is an h3 under ## Shuffle, but spark.comet.exec.transitionRevert.* counts columnar-to-row transitions across a whole stage and is not shuffle specific. The maxTransitions doc string mentions columnar shuffle only as a proxy for why each C2R implies a matching R2C. Promoting this to its own h2 would keep the ## Shuffle section scoped to shuffle.
| By default, Spark compresses shuffle files using LZ4 compression. Comet overrides this behavior with ZSTD compression. | ||
| Compression can be disabled by setting `spark.shuffle.compress=false`, which may result in faster shuffle times in | ||
| certain environments, such as single-node setups with fast NVMe drives, at the expense of increased disk space usage. | ||
| By default, Comet's native shuffle compresses shuffle files with LZ4. Compression can be disabled by setting |
There was a problem hiding this comment.
Both paragraphs scope compression to native shuffle, which leaves the columnar (JVM) shuffle path unaddressed even though the guide documents it above at line 199. spark.comet.exec.shuffle.compression.codec is read by CometShuffleExternalSorter.java:137 and CometDiskBlockWriter.java:145, so it governs columnar shuffle too. And spark.shuffle.compress is consulted only in CometNativeShuffleWriter.scala:223, so setting it to false does not disable compression for columnar shuffle. Could you say which knob applies to which writer? Replacing the old ZSTD claim was the right call and this would finish the correction.
|
|
||
| ## Iceberg Scan Tuning | ||
|
|
||
| When using the native Iceberg scan (`spark.comet.scan.icebergNative.enabled=true`), each task reads its |
There was a problem hiding this comment.
COMET_ICEBERG_NATIVE_ENABLED is createWithDefault(true), so spark.comet.scan.icebergNative.enabled=true reads here as an opt-in when it is already the default. The rest of this section matches the source: dataFileConcurrencyLimit defaults to 1 and its doc string suggests 2 to 8.
|
Thanks for the thorough review, @mbutrovich! I merged main and pushed fixes for all seven comments in 266f251:
PTAL when you get a chance. |
# Conflicts: # benchmarks/pyspark/run_all_benchmarks.sh # spark/src/test/scala/org/apache/comet/CometConfSuite.scala
Audited all CometConf entries and added documentation for user-facing performance and memory knobs that were undocumented in the tuning guide: batch size, spill disk cap, Parquet parallel I/O and filter pushdown, Iceberg data-file concurrency, shuffle compression codec/level, row/columnar transition revert, and metrics overhead. Also fixed an incorrect claim about the default shuffle compression codec (LZ4, not ZSTD) and moved spark.comet.maxTempDirectorySize into the tuning category so it renders in the Memory & Tuning table.
- Remove ### Parallel I/O subsection: the four spark.comet.parquet.read.parallel.io.* and .mergeRanges* configs were removed in apache#4981 - Fix maxTempDirectorySize scope: limit is per Spark task, not per executor - Move ### Parquet Native Scans (merged in from main) under new ## Parquet Reader Tuning h2 so all Parquet guidance is co-located - Promote ### Reducing Row/Columnar Conversion Overhead to its own h2 since spark.comet.exec.transitionRevert.* counts C2R transitions across the whole stage and is not shuffle-specific - Clarify shuffle compression: spark.comet.exec.shuffle.compression.codec governs both native and JVM columnar shuffle; spark.shuffle.compress=false disables compression only for the native shuffle - Fix Iceberg wording: spark.comet.scan.icebergNative.enabled defaults to true
266f251 to
d757c44
Compare
mbutrovich
left a comment
There was a problem hiding this comment.
LGTM pending #4986, which I'll go review now. Thanks @andygrove!
# Conflicts: # docs/source/user-guide/latest/tuning.md
…#4908) * refactor: unify shuffle configs under `spark.comet.shuffle.*` prefix Shuffle-related configs were spread across four disjoint prefixes: `spark.comet.exec.shuffle.*`, `spark.comet.columnar.shuffle.*`, `spark.comet.native.shuffle.*`, and `spark.comet.shuffle.*`. This consolidates all shuffle configs under `spark.comet.shuffle.*` with `.columnar.` and `.native.` sub-namespaces for mode-specific settings. Every renamed key is registered via `.withAlternative(...)`, so existing user configurations continue to work (a deprecation warning is logged when an old key is read). Per-mode leaves that were previously misplaced move to their honest namespace: - `spark.comet.exec.shuffle.writeBufferSize` (native-only) becomes `spark.comet.shuffle.native.writeBufferSize`. - `spark.comet.shuffle.preferDictionary.ratio` (JVM-only) becomes `spark.comet.shuffle.columnar.preferDictionary.ratio`. The `spark.comet.columnar.shuffle.*` keys also fix Category 2 in apache#4978 (dots-in-segment): `spill.threshold`, `memory.factor`, and `batch.size` become `spillThreshold`, `memoryFactor`, and `batchSize`. Part of apache#4978. * refactor: rename `.columnar.` sub-namespace to `.jvm.` The JVM-side shuffle path in Comet is not the only columnar shuffle — native shuffle is also columnar (both use Arrow). The distinguishing axis is *where* the shuffle logic runs. Renaming the sub-namespace to `.jvm.` aligns it with the mode value users already set: `spark.comet.shuffle.mode = jvm`. Keys renamed: - spark.comet.shuffle.columnar.spillThreshold → spark.comet.shuffle.jvm.spillThreshold - spark.comet.shuffle.columnar.memoryFactor → spark.comet.shuffle.jvm.memoryFactor - spark.comet.shuffle.columnar.batchSize → spark.comet.shuffle.jvm.batchSize - spark.comet.shuffle.columnar.preferDictionary.ratio → spark.comet.shuffle.jvm.preferDictionary.ratio The pre-existing deprecated aliases (`spark.comet.columnar.shuffle.*`) are untouched — they were never `.columnar.`-scoped under the new prefix. Val names follow: `COMET_SHUFFLE_COLUMNAR_*` → `COMET_SHUFFLE_JVM_*`. The Scala/Java class names (`CometColumnarShuffle`, `CometColumnarShuffleSuite`, etc.) still use "columnar" and are left alone — that is a separate rename to consider once the config vocabulary settles. * chore: re-run prettier on shuffle docs Table column alignment tweaks after the `.columnar.` → `.jvm.` key rename. Content unchanged. * style: apply spotless formatting to CometShuffleExternalSorter * refactor: move maxBufferBytes under spark.comet.shuffle.native prefix * test: update Spark diffs for renamed shuffle configs Regenerate the Spark SQL test diffs to reference shuffle configs by their string keys rather than CometConf constants, so the config rename in this PR no longer breaks test compilation. - spark.comet.exec.shuffle.enabled -> spark.comet.shuffle.enabled - CometConf.COMET_EXEC_SHUFFLE_WITH_RANGE_PARTITIONING_ENABLED.key -> "spark.comet.shuffle.native.partitioning.range.enabled" * docs: expand tuning guide with performance and memory configs for 1.0 Audited all CometConf entries and added documentation for user-facing performance and memory knobs that were undocumented in the tuning guide: batch size, spill disk cap, Parquet parallel I/O and filter pushdown, Iceberg data-file concurrency, shuffle compression codec/level, row/columnar transition revert, and metrics overhead. Also fixed an incorrect claim about the default shuffle compression codec (LZ4, not ZSTD) and moved spark.comet.maxTempDirectorySize into the tuning category so it renders in the Memory & Tuning table. * style: apply spotless formatting to COMET_MAX_TEMP_DIRECTORY_SIZE doc string * docs: address review feedback on tuning guide - Remove ### Parallel I/O subsection: the four spark.comet.parquet.read.parallel.io.* and .mergeRanges* configs were removed in apache#4981 - Fix maxTempDirectorySize scope: limit is per Spark task, not per executor - Move ### Parquet Native Scans (merged in from main) under new ## Parquet Reader Tuning h2 so all Parquet guidance is co-located - Promote ### Reducing Row/Columnar Conversion Overhead to its own h2 since spark.comet.exec.transitionRevert.* counts C2R transitions across the whole stage and is not shuffle-specific - Clarify shuffle compression: spark.comet.exec.shuffle.compression.codec governs both native and JVM columnar shuffle; spark.shuffle.compress=false disables compression only for the native shuffle - Fix Iceberg wording: spark.comet.scan.icebergNative.enabled defaults to true * docs: use renamed shuffle config keys in tuning guide --------- Co-authored-by: Matt Butrovich <mbutrovich@users.noreply.github.com>
Summary
Audited all
CometConfentries in preparation for the 1.0 release and expanded the Tuning Guide with the user-facing performance and memory knobs that were previously undocumented.New sections in
tuning.md:spark.comet.batchSize(memory vs. throughput tradeoff) and its shuffle-writer cousinspark.comet.shuffle.jvm.batchSizespark.comet.maxTempDirectorySize(100 GB default)spark.comet.scan.icebergNative.dataFileConcurrencyLimitspark.comet.exec.transitionRevert.{enabled,maxTransitions}spark.comet.metrics.enabledis off by defaultAlso:
lz4).spark.comet.maxTempDirectorySizefrom theexeccategory totuningso it renders in the Memory & Tuning table inconfigs.md, and expanded its doc string.Test plan
configs.mdstill generates cleanly (moved config appears in Memory & Tuning table)