docs/source/user-guide/tuning.md - datafusion-comet - Git at Google

 <!---
 Licensed to the Apache Software Foundation (ASF) under one
 or more contributor license agreements.  See the NOTICE file
 distributed with this work for additional information
 regarding copyright ownership.  The ASF licenses this file
 to you under the Apache License, Version 2.0 (the
 "License"); you may not use this file except in compliance
 with the License.  You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing,
 software distributed under the License is distributed on an
 "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 KIND, either express or implied.  See the License for the
 specific language governing permissions and limitations
 under the License.
 -->

 # Tuning Guide

 Comet provides some tuning options to help you get the best performance from your queries.

 ## Memory Tuning

 ### Unified Memory Management with Off-Heap Memory

 The recommended way to share memory between Spark and Comet is to set `spark.memory.offHeap.enabled=true`. This allows
 Comet to share an off-heap memory pool with Spark. The size of the pool is specified by `spark.memory.offHeap.size`. For more details about Spark off-heap memory mode, please refer to Spark documentation: https://spark.apache.org/docs/latest/configuration.html.

 The type of pool can be specified with `spark.comet.exec.memoryPool`.

 The valid pool types are:

 - `unified` (default when `spark.memory.offHeap.enabled=true` is set)
 - `fair_unified`

 The `unified` pool type implements a greedy first-come first-serve limit. This pool works well for queries that do not
 need to spill or have a single spillable operator.

 The `fair_unified` pool type prevents operators from using more than an even fraction of the available memory
 (i.e. `pool_size / num_reservations`). This pool works best when you know beforehand
 the query has multiple operators that will likely all need to spill. Sometimes it will cause spills even
 when there is sufficient memory in order to leave enough memory for other operators.

 ### Dedicated Comet Memory Pools

 Spark uses on-heap memory mode by default, i.e., the `spark.memory.offHeap.enabled` setting is not enabled. If Spark is under on-heap memory mode, Comet will use its own dedicated memory pools that
 are not shared with Spark. This requires additional configuration settings to be specified to set the size and type of
 memory pool to use.

 The size of the pool can be set explicitly with `spark.comet.memoryOverhead`. If this setting is not specified then
 the memory overhead will be calculated by multiplying the executor memory by `spark.comet.memory.overhead.factor`
 (defaults to `0.2`).

 The type of pool can be specified with `spark.comet.exec.memoryPool`. The default setting is `greedy_task_shared`.

 The valid pool types are:

 - `greedy`
 - `greedy_global`
 - `greedy_task_shared`
 - `fair_spill`
 - `fair_spill_global`
 - `fair_spill_task_shared`
 - `unbounded`

 Pool types ending with `_global` use a single global memory pool between all tasks on same executor.

 Pool types ending with `_task_shared` share a single memory pool across all attempts for a single task.

 Other pool types create a dedicated pool per native query plan using a fraction of the available pool size based on number of cores
 and cores per task.

 The `greedy*` pool types use DataFusion's [GreedyMemoryPool], which implements a greedy first-come first-serve limit. This
 pool works well for queries that do not need to spill or have a single spillable operator.

 The `fair_spill*` pool types use DataFusion's [FairSpillPool], which prevents spillable reservations from using more
 than an even fraction of the available memory sans any unspillable reservations
 (i.e. `(pool_size - unspillable_memory) / num_spillable_reservations`). This pool works best when you know beforehand
 the query has multiple spillable operators that will likely all need to spill. Sometimes it will cause spills even
 when there was sufficient memory (reserved for other operators) to avoid doing so. Unspillable memory is allocated in
 a first-come, first-serve fashion

 The `unbounded` pool type uses DataFusion's [UnboundedMemoryPool], which enforces no limit. This option is useful for
 development/testing purposes, where there is no room to allow spilling and rather choose to fail the job.
 Spilling significantly slows down the job and this option is one way to measure the best performance scenario without
 adjusting how much memory to allocate.

 [GreedyMemoryPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.GreedyMemoryPool.html
 [FairSpillPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.FairSpillPool.html
 [UnboundedMemoryPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.UnboundedMemoryPool.html


 ### Determining How Much Memory to Allocate

 Generally, increasing memory overhead will improve query performance, especially for queries containing joins and
 aggregates.

 Once a memory pool is exhausted, the native plan will start spilling to disk, which will slow down the query.

 Insufficient memory allocation can also lead to out-of-memory (OOM) errors.

 ## Configuring spark.executor.memoryOverhead

 In some environments, such as Kubernetes and YARN, it is important to correctly set `spark.executor.memoryOverhead` so
 that it is possible to allocate off-heap memory.

 Comet will automatically set `spark.executor.memoryOverhead` based on the `spark.comet.memory*` settings so that
 resource managers respect Apache Spark memory configuration before starting the containers.

 Note that there is currently a known issue where this will be inaccurate when using Native Memory Management because it
 does not take executor concurrency into account. The tracking issue for this is
 https://github.com/apache/datafusion-comet/issues/949.

 ## Optimizing Joins

 Spark often chooses `SortMergeJoin` over `ShuffledHashJoin` for stability reasons. If the build-side of a
 `ShuffledHashJoin` is very large then it could lead to OOM in Spark.

 Vectorized query engines tend to perform better with `ShuffledHashJoin`, so for best performance it is often preferable
 to configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`. Comet does not yet provide spill-to-disk for
 `ShuffledHashJoin` so this could result in OOM. Also, `SortMergeJoin` may still be faster in some cases. It is best
 to test with both for your specific workloads.

 To configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`, set `spark.comet.exec.replaceSortMergeJoin=true`.

 ## Shuffle

 Comet provides accelerated shuffle implementations that can be used to improve the performance of your queries.

 To enable Comet shuffle, set the following configuration in your Spark configuration:

 ```
 spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager
 spark.comet.exec.shuffle.enabled=true
 ```

 `spark.shuffle.manager` is a Spark static configuration which cannot be changed at runtime.
 It must be set before the Spark context is created. You can enable or disable Comet shuffle
 at runtime by setting `spark.comet.exec.shuffle.enabled` to `true` or `false`.
 Once it is disabled, Comet will fall back to the default Spark shuffle manager.

 ### Shuffle Mode

 Comet provides three shuffle modes: Columnar Shuffle, Native Shuffle and Auto Mode.

 #### Auto Mode

 `spark.comet.exec.shuffle.mode` to `auto` will let Comet choose the best shuffle mode based on the query plan. This
 is the default.

 #### Columnar (JVM) Shuffle

 Comet Columnar shuffle is JVM-based and supports `HashPartitioning`, `RoundRobinPartitioning`, `RangePartitioning`, and
 `SinglePartitioning`. This mode has the highest query coverage.

 Columnar shuffle can be enabled by setting `spark.comet.exec.shuffle.mode` to `jvm`. If this mode is explicitly set,
 then any shuffle operations that cannot be supported in this mode will fall back to Spark.

 #### Native Shuffle

 Comet also provides a fully native shuffle implementation, which generally provides the best performance. However,
 native shuffle currently only supports `HashPartitioning` and `SinglePartitioning`.

 To enable native shuffle, set `spark.comet.exec.shuffle.mode` to `native`. If this mode is explicitly set,
 then any shuffle operations that cannot be supported in this mode will fall back to Spark.

 ### 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.

 ## Explain Plan

 ### Extended Explain

 With Spark 4.0.0 and newer, Comet can provide extended explain plan information in the Spark UI. Currently this lists
 reasons why Comet may not have been enabled for specific operations.
 To enable this, in the Spark configuration, set the following:

 ```shell
 -c spark.sql.extendedExplainProviders=org.apache.comet.ExtendedExplainInfo
 ```

 This will add a section to the detailed plan displayed in the Spark SQL UI page.
	<!---
	Licensed to the Apache Software Foundation (ASF) under one
	or more contributor license agreements. See the NOTICE file
	distributed with this work for additional information
	regarding copyright ownership. The ASF licenses this file
	to you under the Apache License, Version 2.0 (the
	"License"); you may not use this file except in compliance
	with the License. You may obtain a copy of the License at

	http://www.apache.org/licenses/LICENSE-2.0

	Unless required by applicable law or agreed to in writing,
	software distributed under the License is distributed on an
	"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	KIND, either express or implied. See the License for the
	specific language governing permissions and limitations
	under the License.
	-->

	# Tuning Guide

	Comet provides some tuning options to help you get the best performance from your queries.

	## Memory Tuning

	### Unified Memory Management with Off-Heap Memory

	The recommended way to share memory between Spark and Comet is to set `spark.memory.offHeap.enabled=true`. This allows
	Comet to share an off-heap memory pool with Spark. The size of the pool is specified by `spark.memory.offHeap.size`. For more details about Spark off-heap memory mode, please refer to Spark documentation: https://spark.apache.org/docs/latest/configuration.html.

	The type of pool can be specified with `spark.comet.exec.memoryPool`.

	The valid pool types are:

	- `unified` (default when `spark.memory.offHeap.enabled=true` is set)
	- `fair_unified`

	The `unified` pool type implements a greedy first-come first-serve limit. This pool works well for queries that do not
	need to spill or have a single spillable operator.

	The `fair_unified` pool type prevents operators from using more than an even fraction of the available memory
	(i.e. `pool_size / num_reservations`). This pool works best when you know beforehand
	the query has multiple operators that will likely all need to spill. Sometimes it will cause spills even
	when there is sufficient memory in order to leave enough memory for other operators.

	### Dedicated Comet Memory Pools

	Spark uses on-heap memory mode by default, i.e., the `spark.memory.offHeap.enabled` setting is not enabled. If Spark is under on-heap memory mode, Comet will use its own dedicated memory pools that
	are not shared with Spark. This requires additional configuration settings to be specified to set the size and type of
	memory pool to use.

	The size of the pool can be set explicitly with `spark.comet.memoryOverhead`. If this setting is not specified then
	the memory overhead will be calculated by multiplying the executor memory by `spark.comet.memory.overhead.factor`
	(defaults to `0.2`).

	The type of pool can be specified with `spark.comet.exec.memoryPool`. The default setting is `greedy_task_shared`.

	The valid pool types are:

	- `greedy`
	- `greedy_global`
	- `greedy_task_shared`
	- `fair_spill`
	- `fair_spill_global`
	- `fair_spill_task_shared`
	- `unbounded`

	Pool types ending with `_global` use a single global memory pool between all tasks on same executor.

	Pool types ending with `_task_shared` share a single memory pool across all attempts for a single task.

	Other pool types create a dedicated pool per native query plan using a fraction of the available pool size based on number of cores
	and cores per task.

	The `greedy*` pool types use DataFusion's [GreedyMemoryPool], which implements a greedy first-come first-serve limit. This
	pool works well for queries that do not need to spill or have a single spillable operator.

	The `fair_spill*` pool types use DataFusion's [FairSpillPool], which prevents spillable reservations from using more
	than an even fraction of the available memory sans any unspillable reservations
	(i.e. `(pool_size - unspillable_memory) / num_spillable_reservations`). This pool works best when you know beforehand
	the query has multiple spillable operators that will likely all need to spill. Sometimes it will cause spills even
	when there was sufficient memory (reserved for other operators) to avoid doing so. Unspillable memory is allocated in
	a first-come, first-serve fashion

	The `unbounded` pool type uses DataFusion's [UnboundedMemoryPool], which enforces no limit. This option is useful for
	development/testing purposes, where there is no room to allow spilling and rather choose to fail the job.
	Spilling significantly slows down the job and this option is one way to measure the best performance scenario without
	adjusting how much memory to allocate.

	[GreedyMemoryPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.GreedyMemoryPool.html
	[FairSpillPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.FairSpillPool.html
	[UnboundedMemoryPool]: https://docs.rs/datafusion/latest/datafusion/execution/memory_pool/struct.UnboundedMemoryPool.html


	### Determining How Much Memory to Allocate

	Generally, increasing memory overhead will improve query performance, especially for queries containing joins and
	aggregates.

	Once a memory pool is exhausted, the native plan will start spilling to disk, which will slow down the query.

	Insufficient memory allocation can also lead to out-of-memory (OOM) errors.

	## Configuring spark.executor.memoryOverhead

	In some environments, such as Kubernetes and YARN, it is important to correctly set `spark.executor.memoryOverhead` so
	that it is possible to allocate off-heap memory.

	Comet will automatically set `spark.executor.memoryOverhead` based on the `spark.comet.memory*` settings so that
	resource managers respect Apache Spark memory configuration before starting the containers.

	Note that there is currently a known issue where this will be inaccurate when using Native Memory Management because it
	does not take executor concurrency into account. The tracking issue for this is
	https://github.com/apache/datafusion-comet/issues/949.

	## Optimizing Joins

	Spark often chooses `SortMergeJoin` over `ShuffledHashJoin` for stability reasons. If the build-side of a
	`ShuffledHashJoin` is very large then it could lead to OOM in Spark.

	Vectorized query engines tend to perform better with `ShuffledHashJoin`, so for best performance it is often preferable
	to configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`. Comet does not yet provide spill-to-disk for
	`ShuffledHashJoin` so this could result in OOM. Also, `SortMergeJoin` may still be faster in some cases. It is best
	to test with both for your specific workloads.

	To configure Comet to convert `SortMergeJoin` to `ShuffledHashJoin`, set `spark.comet.exec.replaceSortMergeJoin=true`.

	## Shuffle

	Comet provides accelerated shuffle implementations that can be used to improve the performance of your queries.

	To enable Comet shuffle, set the following configuration in your Spark configuration:

	```
	spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager
	spark.comet.exec.shuffle.enabled=true
	```

	`spark.shuffle.manager` is a Spark static configuration which cannot be changed at runtime.
	It must be set before the Spark context is created. You can enable or disable Comet shuffle
	at runtime by setting `spark.comet.exec.shuffle.enabled` to `true` or `false`.
	Once it is disabled, Comet will fall back to the default Spark shuffle manager.

	### Shuffle Mode

	Comet provides three shuffle modes: Columnar Shuffle, Native Shuffle and Auto Mode.

	#### Auto Mode

	`spark.comet.exec.shuffle.mode` to `auto` will let Comet choose the best shuffle mode based on the query plan. This
	is the default.

	#### Columnar (JVM) Shuffle

	Comet Columnar shuffle is JVM-based and supports `HashPartitioning`, `RoundRobinPartitioning`, `RangePartitioning`, and
	`SinglePartitioning`. This mode has the highest query coverage.

	Columnar shuffle can be enabled by setting `spark.comet.exec.shuffle.mode` to `jvm`. If this mode is explicitly set,
	then any shuffle operations that cannot be supported in this mode will fall back to Spark.

	#### Native Shuffle

	Comet also provides a fully native shuffle implementation, which generally provides the best performance. However,
	native shuffle currently only supports `HashPartitioning` and `SinglePartitioning`.

	To enable native shuffle, set `spark.comet.exec.shuffle.mode` to `native`. If this mode is explicitly set,
	then any shuffle operations that cannot be supported in this mode will fall back to Spark.

	### 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.

	## Explain Plan

	### Extended Explain

	With Spark 4.0.0 and newer, Comet can provide extended explain plan information in the Spark UI. Currently this lists
	reasons why Comet may not have been enabled for specific operations.
	To enable this, in the Spark configuration, set the following:

	```shell
	-c spark.sql.extendedExplainProviders=org.apache.comet.ExtendedExplainInfo
	```

	This will add a section to the detailed plan displayed in the Spark SQL UI page.