You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/06 20:43:17 UTC
(datafusion-comet) 01/01: feat: add a spill compression codec config, defaulting to lz4 (#6698)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6698-f7eb8aa3abd6d74ae74d0fa0d50d98f4ce65002c
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 6cd5ea0ed84e10087fac3066baa30ebd65c870ce
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Tue Oct 6 20:42:54 2026 +0000
feat: add a spill compression codec config, defaulting to lz4 (#6698)
* feat: allowlist DataFusion configs, starting with spill compression
Add `spark.comet.exec.allowedDataFusionConfigs`, a list of
`spark.comet.datafusion.*` keys that are passed to native execution
even when the testing flag `spark.comet.exec.respectDataFusionConfigs`
is false. The default list holds
`spark.comet.datafusion.execution.spill_compression`, so compressing
DataFusion's spill files no longer needs the testing flag. Spills stay
uncompressed unless that key is set.
Document spill compression on the memory tuning page and add a nested
and wide data section to the tuning guide.
Closes #6467.
* feat: replace the DataFusion conf allowlist with a spill codec config
Replace `spark.comet.exec.allowedDataFusionConfigs` with
`spark.comet.exec.spill.compression.codec`, which sets the codec for
the files that native sorts, aggregations, sort-merge joins, and nested
loop joins spill through DataFusion. It accepts `lz4`, `zstd`, and
`none`, and defaults to `lz4`, the codec Spark uses for its own spill
files by default, so native spills are now compressed by default.
The JVM sends the resolved value, and the native side maps it to
DataFusion's `SpillCompression` before the `spark.comet.datafusion.*`
testing pass-through, which still wins.
Add a test that a spilling native sort writes fewer bytes with the
default codec than with `none`, and list nested loop joins in the spill
compression docs.
---
docs/source/user-guide/latest/tuning.md | 15 ++++++++
docs/source/user-guide/latest/tuning/memory.md | 10 +++++
native/core/src/execution/jni_api.rs | 25 ++++++++++--
native/core/src/execution/spark_config.rs | 2 +
.../main/scala/org/apache/comet/CometConf.scala | 13 +++++++
.../scala/org/apache/comet/CometExecIterator.scala | 1 +
.../org/apache/comet/exec/CometExecSuite.scala | 45 ++++++++++++++++++++++
7 files changed, 108 insertions(+), 3 deletions(-)
diff --git a/docs/source/user-guide/latest/tuning.md b/docs/source/user-guide/latest/tuning.md
index 158883bb6c..28f6ce112a 100644
--- a/docs/source/user-guide/latest/tuning.md
+++ b/docs/source/user-guide/latest/tuning.md
@@ -36,6 +36,21 @@ guide is split into the following pages:
- [Reducing Row/Columnar Conversion Overhead](tuning/transitions.md): stages in which many
operators fall back to Spark.
+## Nested and Wide Data
+
+Rows with many columns, long strings, or nested columns such as arrays of structs make every batch
+larger, so queries over such data spill more. For these workloads:
+
+- Choose the codec for spill files from native sorts, aggregations, and joins, which Comet
+ compresses with `lz4` by default. See
+ [Compressing Spill Files](tuning/memory.md#compressing-spill-files).
+- Lower the batch size if wide batches cause spilling or out-of-memory errors. See
+ [Batch Size](tuning/memory.md#batch-size).
+- Check which shuffles can run natively when partitioning keys are nested. See
+ [Native Shuffle](tuning/shuffle.md#native-shuffle).
+- Watch for row/columnar conversions, which are expensive for such schemas. See
+ [Wide or Deeply Nested Schemas](tuning/transitions.md#wide-or-deeply-nested-schemas).
+
## Configuring Tokio Runtime
Comet uses a global tokio runtime per executor process. By default it starts one worker thread per executor core
diff --git a/docs/source/user-guide/latest/tuning/memory.md b/docs/source/user-guide/latest/tuning/memory.md
index 1f6ea98a2b..2632369df2 100644
--- a/docs/source/user-guide/latest/tuning/memory.md
+++ b/docs/source/user-guide/latest/tuning/memory.md
@@ -282,3 +282,13 @@ use more than `N` times this value on shared local disks. If the limit is reache
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, remembering that the total across an executor is a
multiple of this value.
+
+## Compressing Spill Files
+
+`spark.comet.exec.spill.compression.codec` sets the codec for the files that native sorts,
+aggregations, sort-merge joins, and nested loop joins spill to disk. The default is `lz4`, which
+Spark also uses for its own spill files by default. `zstd` writes smaller files, which reduces
+spill I/O and how much of `spark.comet.maxTempDirectorySize` they use, but takes more CPU to
+compress and decompress them. `none` writes them uncompressed. Comet's shuffle writers use
+`spark.comet.shuffle.compression.codec` instead. See
+[Shuffle Compression](shuffle.md#shuffle-compression).
diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index 361d3847da..98baf952ef 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -40,6 +40,7 @@ use std::collections::HashSet;
use arrow::array::{Array, RecordBatch};
use arrow::datatypes::DataType as ArrowDataType;
use datafusion::common::{DataFusionError, Result as DataFusionResult};
+use datafusion::config::SpillCompression;
use datafusion::execution::disk_manager::DiskManagerMode;
use datafusion::execution::memory_pool::MemoryPool;
use datafusion::execution::runtime_env::RuntimeEnvBuilder;
@@ -129,9 +130,9 @@ use crate::execution::tracing::{
use crate::execution::memory_pools::logging_pool::LoggingMemoryPool;
use crate::execution::spark_config::{
SparkConfig, COMET_DEBUG_ENABLED, COMET_DEBUG_MEMORY,
- COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED, COMET_EXPLAIN_NATIVE_ENABLED,
- COMET_MAX_TEMP_DIRECTORY_SIZE, COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED,
- COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES,
+ COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED, COMET_EXEC_SPILL_COMPRESSION_CODEC,
+ COMET_EXPLAIN_NATIVE_ENABLED, COMET_MAX_TEMP_DIRECTORY_SIZE,
+ COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED, COMET_TRACING_ENABLED, SPARK_EXECUTOR_CORES,
};
use crate::parquet::encryption_support::{CometEncryptionFactory, ENCRYPTION_FACTORY_ID};
use crate::parquet::parquet_support::CometObjectStoreRegistry;
@@ -902,6 +903,24 @@ fn prepare_datafusion_session_context(
session_config.set_str("datafusion.execution.parquet.reorder_filters", "true");
}
+ // Translate Comet's spill codec into DataFusion's spill compression, which DataFusion
+ // otherwise leaves uncompressed. Set before the pass-through below for the same reason,
+ // so an explicit `spark.comet.datafusion.execution.spill_compression` still wins.
+ let spill_compression = match spark_config
+ .get(COMET_EXEC_SPILL_COMPRESSION_CODEC)
+ .map(String::as_str)
+ {
+ None | Some("lz4") => SpillCompression::Lz4Frame,
+ Some("zstd") => SpillCompression::Zstd,
+ Some("none") => SpillCompression::Uncompressed,
+ Some(codec) => {
+ return Err(CometError::Config(format!(
+ "Unsupported spill compression codec: {codec}"
+ )))
+ }
+ };
+ session_config = session_config.with_spill_compression(spill_compression);
+
// Pass through DataFusion configs from Spark.
// e.g: spark-shell --conf spark.comet.datafusion.sql_parser.parse_float_as_decimal=true
// becomes datafusion.sql_parser.parse_float_as_decimal=true
diff --git a/native/core/src/execution/spark_config.rs b/native/core/src/execution/spark_config.rs
index d29bf6f683..1de88ccbee 100644
--- a/native/core/src/execution/spark_config.rs
+++ b/native/core/src/execution/spark_config.rs
@@ -26,6 +26,8 @@ pub(crate) const COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED: &str =
"spark.comet.parquet.rowFilterPushdown.enabled";
pub(crate) const COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED: &str =
"spark.comet.exec.aggregate.skipPartial.enabled";
+pub(crate) const COMET_EXEC_SPILL_COMPRESSION_CODEC: &str =
+ "spark.comet.exec.spill.compression.codec";
pub(crate) const SPARK_EXECUTOR_CORES: &str = "spark.executor.cores";
/// Comet configs read through this trait must be resolved by the JVM first:
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index 6dee9e3c23..3d08c72dba 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -1265,6 +1265,19 @@ object CometConf extends ShimCometConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefault(100L * 1024 * 1024 * 1024) // 100 GB
+ // Used on native side. Check spark_config.rs how the config is used
+ val COMET_EXEC_SPILL_COMPRESSION_CODEC: ConfigEntry[String] =
+ conf(s"$COMET_EXEC_CONFIG_PREFIX.spill.compression.codec")
+ .category(CATEGORY_TUNING)
+ .doc(
+ "The codec used to compress the files that native sorts, aggregations, and joins " +
+ "spill to disk. lz4 and zstd are supported, and none disables compression. zstd " +
+ "writes smaller files than lz4 but uses more CPU. Comet's shuffle writers use " +
+ s"${COMET_SHUFFLE_COMPRESSION_CODEC.key} instead. $TUNING_GUIDE.")
+ .stringConf
+ .checkValues(Set("lz4", "zstd", "none"))
+ .createWithDefault("lz4")
+
val COMET_RESPECT_DATAFUSION_CONFIGS: ConfigEntry[Boolean] =
conf(s"$COMET_EXEC_CONFIG_PREFIX.respectDataFusionConfigs")
.category(CATEGORY_TESTING)
diff --git a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
index 545c01fafd..0f03a6244e 100644
--- a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
+++ b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala
@@ -870,6 +870,7 @@ object CometExecIterator extends Logging {
CometConf.COMET_DEBUG_ENABLED,
CometConf.COMET_DEBUG_MEMORY_ENABLED,
CometConf.COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED,
+ CometConf.COMET_EXEC_SPILL_COMPRESSION_CODEC,
CometConf.COMET_EXPLAIN_NATIVE_ENABLED,
CometConf.COMET_MAX_TEMP_DIRECTORY_SIZE,
CometConf.COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED,
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
index 5071c40cb6..465da3cdd9 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -112,6 +112,51 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("native sort spill files are compressed with the default spill codec") {
+ val numRows = 20000
+ val compressibleValue = "native-sort-spill-compression-" * 8
+ withTempPath { path =>
+ // A single input file keeps all rows in one task so its sort outgrows the tiny memory
+ // pool below and must spill.
+ spark
+ .createDataFrame((0 until numRows).map(i => (i, compressibleValue)))
+ .coalesce(1)
+ .write
+ .parquet(path.getAbsolutePath)
+
+ withParquetTable(path.getAbsolutePath, "tbl") {
+ def sortSpilledBytes(codecConf: (String, String)*): Long = {
+ var spilledBytes = 0L
+ withSQLConf(
+ Seq(
+ CometConf.COMET_BATCH_SIZE.key -> "1024",
+ CometConf.COMET_OFFHEAP_MEMORY_POOL_FRACTION.key -> "0.002",
+ CometConf.COMET_RESPECT_DATAFUSION_CONFIGS.key -> "true",
+ "spark.comet.datafusion.execution.sort_spill_reservation_bytes" -> "65536") ++
+ codecConf: _*) {
+ val sorted = sql("SELECT * FROM tbl").sortWithinPartitions($"_1".desc)
+ assert(sorted.collect().length == numRows)
+ val plan = sorted.queryExecution.executedPlan
+ val sorts = collect(plan) { case sort: CometSortExec => sort }
+ assert(sorts.nonEmpty, s"Expected a native sort:\n$plan")
+ spilledBytes = sorts.map(_.metrics("spilled_bytes").value).sum
+ assert(spilledBytes > 0L, "Native sort did not spill")
+ }
+ spilledBytes
+ }
+
+ // `spilled_bytes` counts what DataFusion wrote to disk, so this shows the default codec
+ // reaching DataFusion's spill writer, not just crossing JNI.
+ val uncompressed =
+ sortSpilledBytes(CometConf.COMET_EXEC_SPILL_COMPRESSION_CODEC.key -> "none")
+ val compressed = sortSpilledBytes()
+ assert(
+ compressed < uncompressed,
+ s"Spilled $compressed bytes with the default codec and $uncompressed with none")
+ }
+ }
+ }
+
test("sample without replacement") {
withParquetTable((0 until 1000).map(i => (i, i + 1)), "tbl") {
val df = sql("SELECT * FROM tbl").sample(withReplacement = false, fraction = 0.3, seed = 42)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org