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