Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 72 additions & 3 deletions docs/source/user-guide/latest/tuning.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down
6 changes: 4 additions & 2 deletions spark/src/main/scala/org/apache/comet/CometConf.scala
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
Loading