From 322371e0e35727083ede5e5ca6d7bd25407082e9 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 13 Jul 2026 11:06:27 -0600 Subject: [PATCH] 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. --- docs/source/user-guide/latest/tuning.md | 75 ++++++++++++++++++- .../scala/org/apache/comet/CometConf.scala | 6 +- 2 files changed, 76 insertions(+), 5 deletions(-) diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md index a394b4a4cb..e338d3ee30 100644 --- a/docs/source/user-guide/latest/tuning.md +++ b/docs/source/user-guide/latest/tuning.md @@ -103,6 +103,53 @@ Comet Performance It may be possible to reduce Comet's memory overhead by reducing batch sizes or increasing number of partitions. +### Batch Size + +Comet processes data in columnar batches. The batch size is controlled by `spark.comet.batchSize` (default +`8192` rows). Larger batches generally improve throughput by amortizing per-batch overhead, but they also +increase peak memory usage — a batch holds all projected columns in Arrow format at once. Reduce this value +if you see frequent spilling or out-of-memory errors on wide tables; increase it (for example to `16384`) on +narrow tables when memory is plentiful. + +`spark.comet.columnar.shuffle.batch.size` controls the batch size used when the JVM columnar shuffle writer +flushes sorted spill files. It must not exceed `spark.comet.batchSize`. + +### 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 +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 + +### Parallel I/O + +Comet's native Parquet reader can issue overlapping range reads within a single file, which is often the +dominant win when reading from object storage (S3, GCS, ADLS). It is enabled by default via +`spark.comet.parquet.read.parallel.io.enabled=true`, with the thread pool sized by +`spark.comet.parquet.read.parallel.io.thread-pool.size` (default `16` threads per executor). If your +executors have fewer cores or you are reading from local disk, lower this value; if you are reading many +small files from high-latency storage, raise it. When multiple ranges are close together, Comet coalesces +them (`spark.comet.parquet.read.io.mergeRanges`, delta `spark.comet.parquet.read.io.mergeRanges.delta`, +default 8 MB) to reduce request count on cloud storage. + +### Filter Pushdown / Late Materialization + +Setting `spark.comet.parquet.rowFilterPushdown.enabled=true` pushes filter evaluation into the Parquet +decode step and lazily materializes projected columns for surviving rows. This can significantly reduce +CPU and memory when the filter is highly selective on a small subset of columns. It is disabled by default +because it can hurt when the filter is not selective or when most columns must be read anyway. Row-group, +page-index, and bloom-filter pruning happen regardless of this flag whenever Spark's +`spark.sql.parquet.filterPushdown` is on. + +## Iceberg Scan Tuning + +When using the native Iceberg scan (`spark.comet.scan.icebergNative.enabled=true`), each task reads its +data files one at a time by default. For tables with many small files or high-latency storage, increase +`spark.comet.scan.icebergNative.dataFileConcurrencyLimit` (values of 2–8 are suggested) to overlap I/O +across files at the cost of extra memory. + ## Optimizing Sorting on Floating-Point Values Sorting on floating-point data types (or complex types containing floating-point values) is not compatible with @@ -182,9 +229,31 @@ even when both its parent and child are non-Comet operators. ### Shuffle Compression -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 +`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. + +The codec used by Comet's native shuffle is controlled by `spark.comet.exec.shuffle.compression.codec`. Supported +values are `lz4` (default), `zstd`, and `snappy`. LZ4 favors CPU efficiency; ZSTD produces smaller shuffle files +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 + +When a query stage contains many operators that fall back to Spark row-based execution, Comet may insert +repeated columnar-to-row and row-to-columnar conversions that dominate stage runtime. Set +`spark.comet.exec.transitionRevert.enabled=true` to have Comet revert the entire stage to Spark row execution +when the number of columnar-to-row transitions exceeds +`spark.comet.exec.transitionRevert.maxTransitions` (default `2`). This trades native execution of a small +subset of operators for eliminating conversion overhead across the stage. + +## Metrics Overhead + +Comet exposes rich native operator metrics for observability (see [Metrics](metrics.md)), but they are +disabled by default because traversing the Spark plan on every task adds measurable overhead, and metrics +require an external sink (for example Prometheus) to be useful. Enable them with +`spark.comet.metrics.enabled=true` when you have a metrics sink configured. This setting must be applied +before the `SparkSession` is created. ## Explain Plan diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala index 5c130d457e..63edb425d0 100644 --- a/spark/src/main/scala/org/apache/comet/CometConf.scala +++ b/spark/src/main/scala/org/apache/comet/CometConf.scala @@ -869,8 +869,10 @@ object CometConf extends ShimCometConf { // Used on native side. Check spark_config.rs how the config is used val COMET_MAX_TEMP_DIRECTORY_SIZE: ConfigEntry[Long] = conf("spark.comet.maxTempDirectorySize") - .category(CATEGORY_EXEC) - .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 " + + "used by native operators when spilling. Once the limit is reached, further spills " + + "will fail and the query will error out.") .bytesConf(ByteUnit.BYTE) .createWithDefault(100L * 1024 * 1024 * 1024) // 100 GB