mbutrovich commented on code in PR #4908:
URL: https://github.com/apache/datafusion-comet/pull/4908#discussion_r3660656119
##########
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
Review Comment:
This subsection documents four configs that no longer exist on main.
`spark.comet.parquet.read.parallel.io.enabled`,
`spark.comet.parquet.read.parallel.io.thread-pool.size`,
`spark.comet.parquet.read.io.mergeRanges`, and
`spark.comet.parquet.read.io.mergeRanges.delta` were removed in b01655526
(#4981), which merged on 2026-07-20, a week after this branch was cut. Grepping
`spark/`, `common/`, and `native/` on main finds no references to any of them
outside generated `target/` output. #4981 removed them specifically because
they misled operators tuning benchmarks, so documenting them again would
reverse that. Could you drop `### Parallel I/O` entirely?
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -869,8 +869,11 @@ 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 " +
Review Comment:
Same scope question as `tuning.md:120`. The expanded doc string is clearer
than what it replaces, and adding that the limit applies per task rather than
per executor would make it complete, since this string is what shows up in the
generated `configs.md` table.
##########
docs/source/user-guide/latest/tuning.md:
##########
@@ -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
Review Comment:
This is an h3 under `## Shuffle`, but `spark.comet.exec.transitionRevert.*`
counts columnar-to-row transitions across a whole stage and is not shuffle
specific. The `maxTransitions` doc string mentions columnar shuffle only as a
proxy for why each C2R implies a matching R2C. Promoting this to its own h2
would keep the `## Shuffle` section scoped to shuffle.
##########
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
Review Comment:
Once this is rebased, main already has a `### Parquet Native Scans` section
(the row group split mismatch and issue #3817) currently nested under `##
Shuffle` at `tuning.md:189`. That leaves Parquet guidance in two places with
one of them under the wrong parent. Would you move that section under this new
`## Parquet Reader Tuning` heading as part of the rebase?
##########
docs/source/user-guide/latest/tuning.md:
##########
@@ -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
Review Comment:
Both paragraphs scope compression to native shuffle, which leaves the
columnar (JVM) shuffle path unaddressed even though the guide documents it
above at line 199. `spark.comet.exec.shuffle.compression.codec` is read by
`CometShuffleExternalSorter.java:137` and `CometDiskBlockWriter.java:145`, so
it governs columnar shuffle too. And `spark.shuffle.compress` is consulted only
in `CometNativeShuffleWriter.scala:223`, so setting it to `false` does not
disable compression for columnar shuffle. Could you say which knob applies to
which writer? Replacing the old ZSTD claim was the right call and this would
finish the correction.
##########
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
Review Comment:
Is "per executor" the right scope here? The `DiskManagerBuilder` carrying
`with_max_temp_directory_size` is constructed in
`prepare_datafusion_session_context`
(`native/core/src/execution/jni_api.rs:555`), which `createPlan` calls once per
native plan per task, and the comment just below notes the context is scoped to
an executing Spark task. If that reading is right, each task gets its own 100
GB allowance and an executor running N concurrent tasks can reach N times that.
The aggregate is what someone protecting a shared disk needs to reason about,
so stating the per-task scope would make the last sentence of this paragraph
actionable.
##########
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
Review Comment:
`COMET_ICEBERG_NATIVE_ENABLED` is `createWithDefault(true)`, so
`spark.comet.scan.icebergNative.enabled=true` reads here as an opt-in when it
is already the default. The rest of this section matches the source:
`dataFileConcurrencyLimit` defaults to 1 and its doc string suggests 2 to 8.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
