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/09/19 16:08:00 UTC
(datafusion-comet) 01/01: Publish built docs triggered by bce09a64f686325dce5155d2591cfce64e3cd61c
This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 7b3cfe892c0cc0fdf3db06b63e55f432a9795b0b
Author: github-actions[bot] <gi...@users.noreply.github.com>
AuthorDate: Sat Sep 19 16:07:54 2026 +0000
Publish built docs triggered by bce09a64f686325dce5155d2591cfce64e3cd61c
---
_sources/contributor-guide/ffi.md.txt | 8 +-
_sources/contributor-guide/jvm_shuffle.md.txt | 28 +++-
_sources/contributor-guide/native_shuffle.md.txt | 178 +++++++++++++++++---
_sources/user-guide/latest/configs.md.txt | 1 +
_sources/user-guide/latest/installation.md.txt | 10 +-
_sources/user-guide/latest/operators.md.txt | 7 +-
contributor-guide/ffi.html | 7 +-
contributor-guide/jvm_shuffle.html | 21 ++-
contributor-guide/native_shuffle.html | 201 ++++++++++++++++++++---
searchindex.js | 2 +-
user-guide/latest/configs.html | 8 +-
user-guide/latest/installation.html | 9 +-
user-guide/latest/operators.html | 8 +-
13 files changed, 409 insertions(+), 79 deletions(-)
diff --git a/_sources/contributor-guide/ffi.md.txt b/_sources/contributor-guide/ffi.md.txt
index c1f53b138c..e56e5ac0f5 100644
--- a/_sources/contributor-guide/ffi.md.txt
+++ b/_sources/contributor-guide/ffi.md.txt
@@ -100,7 +100,13 @@ implementation depends on the source of the data:
- `ColumnarBatchArrowReader`: an Arrow-backed `ColumnarBatch` (transfers `VectorSchemaRoot` ownership)
The exported `ArrowArrayStream`s are boxed into the `Array[Object]` that `CometExecIterator` / `CometExecRDD` pass
-to native `createPlan` (one slot per scan input; shuffle inputs pass a `CometShuffleBlockIterator` instead).
+to native `createPlan`, one slot per scan input.
+
+Not every slot is an Arrow stream. When a native operator consumes Comet shuffle output and
+`spark.comet.shuffle.directRead.enabled` is set, that slot carries a `CometShuffleBlockIterator` instead, and the
+compressed shuffle blocks are decoded inside the native plan by `ShuffleScanExec` rather than crossing this FFI
+boundary at all. `CometExecRDD.resolveInputObjects` classifies the slots, driven by which scan slots the serialized
+plan marked as `ShuffleScan`. See [Direct Read](native_shuffle.md#direct-read-shufflescan) for that path.
On the native side, `planner.rs` reads each stream's `memoryAddress` and takes ownership through
`AlignedArrowStreamReader::from_raw`, importing the schema once. `ScanExec::get_next_batch` then pulls each batch
diff --git a/_sources/contributor-guide/jvm_shuffle.md.txt b/_sources/contributor-guide/jvm_shuffle.md.txt
index 420278ec49..a4d0b37361 100644
--- a/_sources/contributor-guide/jvm_shuffle.md.txt
+++ b/_sources/contributor-guide/jvm_shuffle.md.txt
@@ -46,10 +46,13 @@ JVM shuffle (`CometColumnarExchange`) is used instead of native shuffle (`CometE
(not a `CometPlan`), JVM shuffle is the only option since native shuffle requires columnar input
from Comet operators.
-3. **Unsupported partition key types**: For `HashPartitioning` and `RangePartitioning`, native shuffle
- only supports primitive types as partition keys. Complex types (struct, array, map) cannot be used
- as partition keys in native shuffle and will fall back to JVM columnar shuffle. Note that complex types are
- fully supported as data columns in both implementations.
+3. **Unsupported partition key types**: `RangePartitioning` keys must be primitive, so a complex
+ range key always falls back here. `HashPartitioning` keys must be primitive only by default:
+ setting `spark.comet.shuffle.native.partitioning.hash.nested.enabled` to `true` keeps struct and
+ array keys, and map keys on Spark 4.0 and later, on the native path. The config defaults to
+ `false`, so a complex hash key falls back to JVM columnar shuffle unless it is enabled. See
+ [Supported partition key types](native_shuffle.md#when-native-shuffle-is-used) for the exact
+ rules. Complex types are fully supported as data columns in both implementations.
## Input Handling
@@ -179,6 +182,12 @@ Selection logic in `CometShuffleManager.shouldBypassMergeSort()`:
3. Native code (`Native.decodeShuffleBlock()`) decompresses and decodes to Arrow arrays
4. Arrow FFI imports arrays as `ColumnarBatch`
+This is the path taken when the consumer is a Spark operator. When a Comet native operator consumes
+the shuffle output and `spark.comet.shuffle.directRead.enabled` is set, steps 2 through 4 are skipped:
+the raw compressed blocks are handed to native code, which decodes them inside the plan. JVM shuffle
+writes the same Arrow IPC block format as native shuffle, so direct read applies to both. See
+[Direct Read](native_shuffle.md#direct-read-shufflescan) for the mechanism.
+
## Memory Management
- `CometShuffleMemoryAllocator`: Custom allocator for off-heap memory pages
@@ -187,8 +196,9 @@ Selection logic in `CometShuffleManager.shouldBypassMergeSort()`:
## Configuration
-| Config | Description |
-| ---------------------------------------- | ----------------------------------- |
-| `spark.comet.shuffle.jvm.batchSize` | Rows per Arrow batch |
-| `spark.comet.shuffle.jvm.spillThreshold` | Row count threshold for spill |
-| `spark.comet.shuffle.compression.codec` | Compression codec (zstd, lz4, etc.) |
+| Config | Description |
+| ---------------------------------------- | ------------------------------------------------- |
+| `spark.comet.shuffle.jvm.batchSize` | Rows per Arrow batch |
+| `spark.comet.shuffle.jvm.spillThreshold` | Row count threshold for spill |
+| `spark.comet.shuffle.compression.codec` | Compression codec, `lz4` by default |
+| `spark.comet.shuffle.directRead.enabled` | Decode blocks natively on read, `true` by default |
diff --git a/_sources/contributor-guide/native_shuffle.md.txt b/_sources/contributor-guide/native_shuffle.md.txt
index 1608b70f46..6d7f4a18fa 100644
--- a/_sources/contributor-guide/native_shuffle.md.txt
+++ b/_sources/contributor-guide/native_shuffle.md.txt
@@ -49,14 +49,26 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
columnar output. Row-based Spark operators require JVM shuffle.
3. **Supported partitioning type**: Native shuffle supports:
+
- `HashPartitioning`
- `RangePartitioning`
- `SinglePartition`
- - `RoundRobinPartitioning`
-
-4. **Supported partition key types**: For `HashPartitioning` and `RangePartitioning`, partition
- keys must be primitive types. Complex types (struct, array, map) as partition keys require
- JVM shuffle. Note that complex types are fully supported as data columns in native shuffle.
+ - `RoundRobinPartitioning`, disabled by default via
+ `spark.comet.shuffle.native.partitioning.roundrobin.enabled`, because Comet's hash-based
+ assignment puts unsorted rows in different partitions than Spark does
+
+4. **Supported partition key types**: The rule differs by partitioning, and neither restricts data
+ columns. Complex types are fully supported as data columns in native shuffle.
+ - `RangePartitioning` keys must be primitive. `supportedRangePartitioningDataType` rejects every
+ nested type, because native cannot sort them, and rejects collated strings, because Comet
+ compares raw bytes. Float and double are also rejected when
+ `spark.comet.exec.strictFloatingPoint` is enabled.
+ - `HashPartitioning` keys must be primitive **by default**. Setting
+ `spark.comet.shuffle.native.partitioning.hash.nested.enabled` to `true` admits structs and
+ arrays as keys, checked recursively to their leaves, and maps on Spark 4.0 and later, where
+ Spark's `mapsort` normalization makes physical entry order irrelevant. A collated string at any
+ depth still disqualifies the key. The config defaults to `false` pending measurement of the
+ nested hashing paths, so by default a complex hash key falls back to JVM shuffle.
## Architecture
@@ -115,11 +127,20 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
### Rust Side
-| File | Location | Description |
-| ----------------------- | ------------------------------------ | ------------------------------------------------------------------------------------ |
-| `shuffle_writer.rs` | `native/core/src/execution/shuffle/` | `ShuffleWriterExec` plan and partitioners. Main shuffle logic. |
-| `codec.rs` | `native/core/src/execution/shuffle/` | `ShuffleBlockWriter` for Arrow IPC encoding with compression. Also handles decoding. |
-| `comet_partitioning.rs` | `native/core/src/execution/shuffle/` | `CometPartitioning` enum defining partition schemes (Hash, Range, Single). |
+The native shuffle implementation is its own workspace crate, `datafusion-comet-shuffle`, rooted at
+`native/shuffle/`. Paths below are relative to `native/shuffle/src/`.
+
+| File | Description |
+| ---------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- |
+| `shuffle_writer.rs` | `ShuffleWriterExec`, the DataFusion `ExecutionPlan` that drives partitioning and writing. |
+| `comet_partitioning.rs` | `CometPartitioning` enum defining the partition schemes (Hash, Range, Single, RoundRobin). |
+| `partitioners/multi_partition.rs` | `MultiPartitionShuffleRepartitioner` for hash, range, and round robin partitioning. |
+| `partitioners/single_partition.rs` | `SinglePartitionShufflePartitioner` for the single partition case. |
+| `writers/shuffle_block_writer.rs` | `ShuffleBlockWriter` and the `CompressionCodec` enum. Arrow IPC encoding with compression. |
+| `writers/partition_writer.rs` | The `PartitionWriter` trait. Implemented by `LocalPartitionWriter` (`writers/local/`) and `RssPartitionWriter` (`writers/rss/`). |
+| `writers/buf_batch_writer.rs` | `BufBatchWriter`, which coalesces sub-`batch_size` batches through Arrow's `BatchCoalescer` before serializing. |
+| `writers/local/spill.rs` | `PartitionedSpill`, the one spill file shared by every output partition, and its per-partition byte ranges. |
+| `ipc.rs` | `read_ipc_compressed` and `read_ipc_compressed_validated`, the decode side used by the shuffle reader. |
## Data Flow
@@ -128,6 +149,7 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
1. **Plan construction**: `CometNativeShuffleWriter` builds a protobuf operator tree with a
`ShuffleWriter` operator at the root and `childNativeOp` as its child. `childNativeOp` takes
one of two shapes:
+
- The child plan's `nativeOp` directly, when `CometShuffleExchangeExec`'s child is a
`CometNativeExec` subtree. The upstream operators run inside the same `CometExecIterator`
as the writer, with no JVM-to-native batch boundary between them.
@@ -140,6 +162,7 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
2. **Native execution**: A single `CometExecIterator` per partition runs the unified plan.
3. **Partitioning**: `ShuffleWriterExec` receives batches and routes to the appropriate partitioner:
+
- `MultiPartitionShuffleRepartitioner`: For hash/range/round-robin partitioning
- `SinglePartitionShufflePartitioner`: For single partition (simpler path)
@@ -147,6 +170,7 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
exceeds the threshold, partitions spill to temporary files.
5. **Encoding**: `ShuffleBlockWriter` encodes each partition's data as compressed Arrow IPC:
+
- Writes compression type header
- Writes field count header
- Writes compressed IPC stream
@@ -160,9 +184,23 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
### Read Path
+There are two read paths, and which one runs is decided at plan serialization time rather than at
+read time. See [Direct Read](#direct-read-shufflescan) below for how the choice is made.
+
+**Direct read**, the default, decodes inside the native plan:
+
+1. `CometBlockStoreShuffleReader.readAsRawStream()` concatenates the fetched block streams and
+ skips decoding entirely.
+2. `CometShuffleBlockIterator` hands native one compressed block at a time.
+3. The native `ShuffleScanExec` decodes each block with `read_ipc_compressed` and feeds the
+ resulting `RecordBatch` straight into the plan. No Arrow FFI export or import happens.
+
+**JVM decode**, used when direct read does not apply:
+
1. `CometBlockStoreShuffleReader` fetches shuffle blocks via `ShuffleBlockFetcherIterator`.
2. For each block, `NativeBatchDecoderIterator`:
+
- Reads the 8-byte compressed length header
- Reads the 8-byte field count header
- Reads the compressed IPC data
@@ -172,6 +210,70 @@ Native shuffle (`CometExchange`) is selected when all of the following condition
4. Arrow FFI transfers the `RecordBatch` to JVM as a `ColumnarBatch`.
+## Direct Read (ShuffleScan)
+
+Direct read lets a native operator consume shuffle output without the batch ever being decoded in
+the JVM or crossing Arrow FFI. It is controlled by `spark.comet.shuffle.directRead.enabled`, which
+defaults to `true` and requires `spark.comet.shuffle.enabled`. It applies to both native shuffle and
+JVM columnar shuffle, because both write the same Arrow IPC block format.
+
+### How the path is selected
+
+`CometExchangeSink.shouldUseShuffleScan` (`spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala`)
+decides during plan serialization. When direct read is enabled and the sink's input is a Comet shuffle
+exchange, `convertToShuffleScan` emits a `ShuffleScan` operator. When either is false the sink falls
+through to the base `CometSink.convert`, which emits the usual `Scan`.
+
+The two are not alternatives on failure. If any output type fails `supportedSinkDataType`,
+`convertToShuffleScan` records the fallback reason `Unsupported data type for shuffle direct read`
+and returns `None`. It does not retry as a regular `Scan`, and retrying would not help, because
+`CometSink.convert` gates on the same `supportedSinkDataType`. `CometExecRule` calls this as
+`convertToComet(s, CometExchangeSink).getOrElse(s)`, so `None` leaves the original Spark shuffle stage
+in the plan. An unsupported output type therefore means the stage is not converted natively at all,
+not that it is served over the FFI read path instead.
+
+**The protobuf is the source of truth for which slots are direct read, not the config.**
+`findShuffleScanIndices` (`operators.scala`) walks the serialized plan, counting scan slots in order
+and collecting the indices that carry a `ShuffleScan`. JVM-side input dispatch reads that set rather
+than re-checking the config, so the two cannot disagree. Anything changing the serde condition has to
+leave that walk consistent with it.
+
+### How blocks reach native
+
+`CometExecRDD.resolveInputObjects` fills one input slot per scan input, in scan-input order:
+
+- A slot in `shuffleScanIndices` gets a `CometShuffleBlockIterator`, obtained from
+ `CometShuffledBatchRDD.computeAsShuffleBlockIterator`. A slot marked as a shuffle scan whose RDD is
+ not a `CometShuffledBatchRDD` throws `CometRuntimeException`.
+- Every other slot gets the `ArrowArrayStream` exported by the ordinary FFI path.
+
+`CometShuffleBlockIterator` reads a 16-byte header per block: an 8-byte compressed length, which
+includes the field count but not itself, followed by an 8-byte field count that is discarded because
+the schema comes from the `ShuffleScan` protobuf fields. The compressed body is read into a reused
+`DirectByteBuffer`, **valid only until the next `hasNext()` call**. Native must fully consume it
+before pulling the next block, which `read_ipc_compressed` satisfies because it allocates fresh
+native memory for the decoded data.
+
+### Native side
+
+`ShuffleScanExec` (`native/core/src/execution/operators/shuffle_scan.rs`) pulls blocks through the
+iterator's `hasNext()` and `getBuffer()` JNI methods and decodes them with `read_ipc_compressed`.
+Two details matter when changing it:
+
+- `get_next_batch` is called from outside `poll_next`, because JNI calls cannot be made from tokio
+ worker threads. A change that moves the JNI call into the stream's `poll_next` breaks this.
+- Dictionary-encoded columns are unpacked to their value type by `unpack_dictionary`, so the schema
+ the plan sees matches the declared `ShuffleScan` fields.
+
+Blocks are validated before decoding when `CometShuffleBlockIterator.requiresValidation()` is true,
+which happens when the underlying stream implements `CometShuffleReadFailureHandler` so a corrupt
+block can be reported back as a shuffle read failure rather than a native decode panic.
+
+Celeborn reuses the same expected-schema contract from the JVM side. `CometCelebornShuffleReader`
+builds a `ShuffleScan` message from the dependency's output attributes and passes it to
+`NativeBatchDecoderIterator`, so remote logical types are validated before Arrow import and an
+unsupported type fails before any shuffle resource is acquired.
+
## Partitioning
### Hash Partitioning
@@ -211,6 +313,11 @@ assignment only provides even distribution when the data has sufficient variatio
hashed columns. Data with low cardinality or identical values may result in skewed partition
sizes.
+Because Spark assigns round robin partitions by sorting rows on their binary `UnsafeRow` form,
+which Arrow's layout does not reproduce, unsorted output can land in different partitions than
+Spark's. Sorted output is identical. That difference is why
+`spark.comet.shuffle.native.partitioning.roundrobin.enabled` defaults to `false`.
+
## Memory Management
Native shuffle uses DataFusion's memory management with spilling support:
@@ -219,15 +326,29 @@ Native shuffle uses DataFusion's memory management with spilling support:
- **Spill triggers**: Partitions spill to disk when the memory pool denies an allocation, or
when the buffered bytes reach `spark.comet.shuffle.native.maxBufferBytes`. That config defaults to
0, which disables the fixed limit and leaves memory pressure as the only trigger.
-- **Per-partition spilling**: Each partition has its own spill file. Multiple spills for a
- partition are concatenated when writing the final output.
+- **One spill file per task**: Every output partition spills into the same file, not one file per
+ partition. `PartitionedSpill` records the byte ranges each partition's blocks occupy, in write
+ order, and the final output copies a partition's ranges back in that order.
- **Scratch space**: Reusable buffers for partition ID computation to reduce allocations.
-The `MultiPartitionShuffleRepartitioner` manages:
-
-- `PartitionBuffer`: In-memory buffer for each partition
-- `SpillFile`: Temporary file for spilled data
-- Memory tracking via `MemoryConsumer` trait
+The `MultiPartitionShuffleRepartitioner` holds:
+
+- `buffered_batches`, a `Vec<RecordBatch>` of incoming batches, alongside `partition_indices`
+ recording which rows of those batches belong to each partition. Rows are not copied into
+ per-partition buffers as they arrive.
+- `reservation`, a `MemoryReservation` charged for the bytes each buffered batch newly pins.
+ `pinned_buffers` tracks backing buffer start addresses so one allocation shared by many sliced
+ batches is charged once rather than once per slice. Charging a per-batch size instead would
+ overstate memory by the slice count and spill spuriously.
+- `max_buffer_bytes`, the optional fixed spill threshold described above. `None` leaves pool
+ pressure as the only trigger.
+- `scratch`, reusable buffers for partition ID computation.
+
+The spill file is owned by `PartitionedSpill` in `writers/local/spill.rs`. It holds one DataFusion
+`SpillFile`, created lazily on the first spill and shared by every output partition, and tracks per
+partition the byte ranges holding that partition's blocks in write order. A partial write sets a
+`failed` flag, because a write that stops midway leaves the recorded ranges unable to describe the
+file. The single-partition writer writes straight to the output file and never spills.
## Compression
@@ -246,14 +367,21 @@ independently compressed, allowing parallel decompression during reads.
## Configuration
-| Config | Default | Description |
-| -------------------------------------------- | ------- | ---------------------------------------- |
-| `spark.comet.shuffle.enabled` | `true` | Enable Comet shuffle |
-| `spark.comet.shuffle.mode` | `auto` | Shuffle mode: `native`, `jvm`, or `auto` |
-| `spark.comet.shuffle.compression.codec` | `zstd` | Compression codec |
-| `spark.comet.shuffle.compression.zstd.level` | `1` | Zstd compression level |
-| `spark.comet.shuffle.native.writeBufferSize` | `1MB` | Write buffer size |
-| `spark.comet.shuffle.jvm.batchSize` | `8192` | Target rows per batch |
+| Config | Default | Description |
+| ------------------------------------------------------------------- | ------- | ------------------------------------------------------------- |
+| `spark.comet.shuffle.enabled` | `true` | Enable Comet shuffle |
+| `spark.comet.shuffle.mode` | `auto` | Shuffle mode: `native`, `jvm`, or `auto` |
+| `spark.comet.shuffle.directRead.enabled` | `true` | Decode shuffle blocks in native code, bypassing Arrow FFI |
+| `spark.comet.shuffle.compression.codec` | `lz4` | Compression codec |
+| `spark.comet.shuffle.compression.zstd.level` | `1` | Zstd compression level |
+| `spark.comet.shuffle.native.writeBufferSize` | `1MB` | Write buffer size |
+| `spark.comet.shuffle.native.maxBufferBytes` | `0` | Fixed spill threshold. `0` disables it, leaving pool pressure |
+| `spark.comet.shuffle.native.partitioning.hash.enabled` | `true` | Allow `HashPartitioning` on the native path |
+| `spark.comet.shuffle.native.partitioning.hash.nested.enabled` | `false` | Allow struct and array hash keys, and map keys on Spark 4.0+ |
+| `spark.comet.shuffle.native.partitioning.range.enabled` | `true` | Allow `RangePartitioning` on the native path |
+| `spark.comet.shuffle.native.partitioning.roundrobin.enabled` | `false` | Allow `RoundRobinPartitioning` on the native path |
+| `spark.comet.shuffle.native.partitioning.roundrobin.maxHashColumns` | `0` | Columns to hash for round robin. `0` hashes all of them |
+| `spark.comet.shuffle.jvm.batchSize` | `8192` | Target rows per batch |
## Comparison with JVM Shuffle
diff --git a/_sources/user-guide/latest/configs.md.txt b/_sources/user-guide/latest/configs.md.txt
index 57b3e5ba6d..0f720d4aa5 100644
--- a/_sources/user-guide/latest/configs.md.txt
+++ b/_sources/user-guide/latest/configs.md.txt
@@ -72,6 +72,7 @@ Comet provides the following configuration settings.
| `spark.comet.metrics.updateInterval` | The interval in milliseconds to update metrics. If interval is negative, metrics will be updated upon task completion. | 3000 |
| `spark.comet.nativeLoadRequired` | Whether to require Comet native library to load successfully when Comet is enabled. If not, Comet will silently fallback to Spark when it fails to load the native lib. Otherwise, an error will be thrown and the Spark job will be aborted. | false |
| `spark.comet.operator.DataWritingCommandExec.allowIncompatible` | Whether to allow incompatibility for operator: DataWritingCommandExec. False by default. Can be overridden with SPARK_COMET_OPERATOR_DATAWRITINGCOMMANDEXEC_ALLOWINCOMPATIBLE env variable It can be overridden by the environment variable `SPARK_COMET_OPERATOR_DATAWRITINGCOMMANDEXEC_ALLOWINCOMPATIBLE`. | false |
+| `spark.comet.operator.WriteFilesExec.allowIncompatible` | Whether to allow incompatibility for operator: WriteFilesExec. False by default. Can be overridden with SPARK_COMET_OPERATOR_WRITEFILESEXEC_ALLOWINCOMPATIBLE env variable It can be overridden by the environment variable `SPARK_COMET_OPERATOR_WRITEFILESEXEC_ALLOWINCOMPATIBLE`. | false |
| `spark.comet.sparkToColumnar.enabled` | Whether to enable Spark to Arrow columnar conversion. When this is turned on, Comet will convert operators in `spark.comet.sparkToColumnar.supportedOperatorList` into Arrow columnar format before processing. | false |
| `spark.comet.sparkToColumnar.supportedOperatorList` | A comma-separated list of operators that will be converted to Arrow columnar format when `spark.comet.sparkToColumnar.enabled` is true. | Range,InMemoryTableScan,RDDScan,OneRowRelation |
<!-- prettier-ignore-end -->
diff --git a/_sources/user-guide/latest/installation.md.txt b/_sources/user-guide/latest/installation.md.txt
index 4bdbe8fa06..f7ee94edd8 100644
--- a/_sources/user-guide/latest/installation.md.txt
+++ b/_sources/user-guide/latest/installation.md.txt
@@ -158,16 +158,20 @@ Create a test Parquet source
scala> (0 until 10).toDF("a").write.mode("overwrite").parquet("/tmp/test")
```
-Comet will log output similar to:
+Comet will log output similar to this on Spark 4.0 and later:
```shell
INFO core/src/lib.rs: Comet native library version 1.1.0-SNAPSHOT initialized
WARN CometExecRule: Comet cannot execute some parts of this plan natively (set spark.comet.explain.fallback.enabled=false to disable this logging):
- Execute InsertIntoHadoopFsRelationCommand [COMET: Native support for operator DataWritingCommandExec is disabled. Set spark.comet.parquet.write.enabled=true to enable it.]
-+- WriteFiles
+ Execute InsertIntoHadoopFsRelationCommand
++- WriteFiles [COMET: Native support for operator WriteFilesExec is disabled. Set spark.comet.parquet.write.enabled=true to enable it.]
+- LocalTableScan [COMET: Native support for operator LocalTableScanExec is disabled. Set spark.comet.exec.localTableScan.enabled=true to enable it.]
```
+On Spark 3.4 and 3.5 the native writer replaces the whole write command rather than just the
+per-task write, so the same message appears on `Execute InsertIntoHadoopFsRelationCommand` and
+names `DataWritingCommandExec`.
+
Query the data from the test source and check:
- INFO message shows the native Comet library has been initialized.
diff --git a/_sources/user-guide/latest/operators.md.txt b/_sources/user-guide/latest/operators.md.txt
index 6a6ea5aa4d..2182e7de58 100644
--- a/_sources/user-guide/latest/operators.md.txt
+++ b/_sources/user-guide/latest/operators.md.txt
@@ -117,9 +117,10 @@ omitted from the tables below and may be reconsidered based on demand:
## Writes
-| Operator | Status | Notes |
-| ------------------------ | ------ | ----------------------------------------------------------------- |
-| `DataWritingCommandExec` | ⚠️ | Experimental native Parquet writes, disabled by default (opt-in). |
+| Operator | Status | Notes |
+| ------------------------ | ------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `WriteFilesExec` | ⚠️ | Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when `spark.sql.files.maxRecordsPerFile` is set. |
+| `DataWritingCommandExec` | ⚠️ | Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by `WriteFilesExec` on Spark 4.0+ and removed with Spark 3.x support. |
## Python and UDF
diff --git a/contributor-guide/ffi.html b/contributor-guide/ffi.html
index 06fb6cd9ac..3514857d6f 100644
--- a/contributor-guide/ffi.html
+++ b/contributor-guide/ffi.html
@@ -629,7 +629,12 @@ implementation depends on the source of the data:</p>
<li><p><code class="docutils literal notranslate"><span class="pre">ColumnarBatchArrowReader</span></code>: an Arrow-backed <code class="docutils literal notranslate"><span class="pre">ColumnarBatch</span></code> (transfers <code class="docutils literal notranslate"><span class="pre">VectorSchemaRoot</span></code> ownership)</p></li>
</ul>
<p>The exported <code class="docutils literal notranslate"><span class="pre">ArrowArrayStream</span></code>s are boxed into the <code class="docutils literal notranslate"><span class="pre">Array[Object]</span></code> that <code class="docutils literal notranslate"><span class="pre">CometExecIterator</span></code> / <code class="docutils literal notranslate"><span class="pre">CometExecRDD</span></code> pass
-to native <code class="docutils literal notranslate"><span class="pre">createPlan</span></code> (one slot per scan input; shuffle inputs pass a <code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator</span></code> instead).</p>
+to native <code class="docutils literal notranslate"><span class="pre">createPlan</span></code>, one slot per scan input.</p>
+<p>Not every slot is an Arrow stream. When a native operator consumes Comet shuffle output and
+<code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.directRead.enabled</span></code> is set, that slot carries a <code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator</span></code> instead, and the
+compressed shuffle blocks are decoded inside the native plan by <code class="docutils literal notranslate"><span class="pre">ShuffleScanExec</span></code> rather than crossing this FFI
+boundary at all. <code class="docutils literal notranslate"><span class="pre">CometExecRDD.resolveInputObjects</span></code> classifies the slots, driven by which scan slots the serialized
+plan marked as <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code>. See <a class="reference internal" href="native_shuffle.html#direct-read-shufflescan"><span class="std std-ref">Direct Read</span></a> for that path.</p>
<p>On the native side, <code class="docutils literal notranslate"><span class="pre">planner.rs</span></code> reads each stream’s <code class="docutils literal notranslate"><span class="pre">memoryAddress</span></code> and takes ownership through
<code class="docutils literal notranslate"><span class="pre">AlignedArrowStreamReader::from_raw</span></code>, importing the schema once. <code class="docutils literal notranslate"><span class="pre">ScanExec::get_next_batch</span></code> then pulls each batch
through the stream’s <code class="docutils literal notranslate"><span class="pre">get_next</span></code> callback. There is no per-batch JNI call and no per-column FFI export.</p>
diff --git a/contributor-guide/jvm_shuffle.html b/contributor-guide/jvm_shuffle.html
index 500ce71425..c27d9b39e7 100644
--- a/contributor-guide/jvm_shuffle.html
+++ b/contributor-guide/jvm_shuffle.html
@@ -572,10 +572,13 @@ Uses Spark’s partitioner for partition assignment. Can accept either row-based
<li><p><strong>Child plan is not a Comet native operator</strong>: When the child plan is a Spark row-based operator
(not a <code class="docutils literal notranslate"><span class="pre">CometPlan</span></code>), JVM shuffle is the only option since native shuffle requires columnar input
from Comet operators.</p></li>
-<li><p><strong>Unsupported partition key types</strong>: For <code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code> and <code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code>, native shuffle
-only supports primitive types as partition keys. Complex types (struct, array, map) cannot be used
-as partition keys in native shuffle and will fall back to JVM columnar shuffle. Note that complex types are
-fully supported as data columns in both implementations.</p></li>
+<li><p><strong>Unsupported partition key types</strong>: <code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code> keys must be primitive, so a complex
+range key always falls back here. <code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code> keys must be primitive only by default:
+setting <code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.hash.nested.enabled</span></code> to <code class="docutils literal notranslate"><span class="pre">true</span></code> keeps struct and
+array keys, and map keys on Spark 4.0 and later, on the native path. The config defaults to
+<code class="docutils literal notranslate"><span class="pre">false</span></code>, so a complex hash key falls back to JVM columnar shuffle unless it is enabled. See
+<a class="reference internal" href="native_shuffle.html#when-native-shuffle-is-used"><span class="std std-ref">Supported partition key types</span></a> for the exact
+rules. Complex types are fully supported as data columns in both implementations.</p></li>
</ol>
</section>
<section id="input-handling">
@@ -774,6 +777,11 @@ to protobuf and implements the partitioning logic in native code.</p>
<li><p>Native code (<code class="docutils literal notranslate"><span class="pre">Native.decodeShuffleBlock()</span></code>) decompresses and decodes to Arrow arrays</p></li>
<li><p>Arrow FFI imports arrays as <code class="docutils literal notranslate"><span class="pre">ColumnarBatch</span></code></p></li>
</ol>
+<p>This is the path taken when the consumer is a Spark operator. When a Comet native operator consumes
+the shuffle output and <code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.directRead.enabled</span></code> is set, steps 2 through 4 are skipped:
+the raw compressed blocks are handed to native code, which decodes them inside the plan. JVM shuffle
+writes the same Arrow IPC block format as native shuffle, so direct read applies to both. See
+<a class="reference internal" href="native_shuffle.html#direct-read-shufflescan"><span class="std std-ref">Direct Read</span></a> for the mechanism.</p>
</section>
</section>
<section id="memory-management">
@@ -800,7 +808,10 @@ to protobuf and implements the partitioning logic in native code.</p>
<td><p>Row count threshold for spill</p></td>
</tr>
<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.compression.codec</span></code></p></td>
-<td><p>Compression codec (zstd, lz4, etc.)</p></td>
+<td><p>Compression codec, <code class="docutils literal notranslate"><span class="pre">lz4</span></code> by default</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.directRead.enabled</span></code></p></td>
+<td><p>Decode blocks natively on read, <code class="docutils literal notranslate"><span class="pre">true</span></code> by default</p></td>
</tr>
</tbody>
</table>
diff --git a/contributor-guide/native_shuffle.html b/contributor-guide/native_shuffle.html
index ed1dc299a6..6803d67f7c 100644
--- a/contributor-guide/native_shuffle.html
+++ b/contributor-guide/native_shuffle.html
@@ -576,12 +576,26 @@ columnar output. Row-based Spark operators require JVM shuffle.</p></li>
<li><p><code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code></p></li>
<li><p><code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code></p></li>
<li><p><code class="docutils literal notranslate"><span class="pre">SinglePartition</span></code></p></li>
-<li><p><code class="docutils literal notranslate"><span class="pre">RoundRobinPartitioning</span></code></p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">RoundRobinPartitioning</span></code>, disabled by default via
+<code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.roundrobin.enabled</span></code>, because Comet’s hash-based
+assignment puts unsorted rows in different partitions than Spark does</p></li>
+</ul>
+</li>
+<li><p><strong>Supported partition key types</strong>: The rule differs by partitioning, and neither restricts data
+columns. Complex types are fully supported as data columns in native shuffle.</p>
+<ul class="simple">
+<li><p><code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code> keys must be primitive. <code class="docutils literal notranslate"><span class="pre">supportedRangePartitioningDataType</span></code> rejects every
+nested type, because native cannot sort them, and rejects collated strings, because Comet
+compares raw bytes. Float and double are also rejected when
+<code class="docutils literal notranslate"><span class="pre">spark.comet.exec.strictFloatingPoint</span></code> is enabled.</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code> keys must be primitive <strong>by default</strong>. Setting
+<code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.hash.nested.enabled</span></code> to <code class="docutils literal notranslate"><span class="pre">true</span></code> admits structs and
+arrays as keys, checked recursively to their leaves, and maps on Spark 4.0 and later, where
+Spark’s <code class="docutils literal notranslate"><span class="pre">mapsort</span></code> normalization makes physical entry order irrelevant. A collated string at any
+depth still disqualifies the key. The config defaults to <code class="docutils literal notranslate"><span class="pre">false</span></code> pending measurement of the
+nested hashing paths, so by default a complex hash key falls back to JVM shuffle.</p></li>
</ul>
</li>
-<li><p><strong>Supported partition key types</strong>: For <code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code> and <code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code>, partition
-keys must be primitive types. Complex types (struct, array, map) as partition keys require
-JVM shuffle. Note that complex types are fully supported as data columns in native shuffle.</p></li>
</ol>
</section>
<section id="architecture">
@@ -668,25 +682,41 @@ JVM shuffle. Note that complex types are fully supported as data columns in nati
</section>
<section id="rust-side">
<h3>Rust Side<a class="headerlink" href="#rust-side" title="Link to this heading">#</a></h3>
+<p>The native shuffle implementation is its own workspace crate, <code class="docutils literal notranslate"><span class="pre">datafusion-comet-shuffle</span></code>, rooted at
+<code class="docutils literal notranslate"><span class="pre">native/shuffle/</span></code>. Paths below are relative to <code class="docutils literal notranslate"><span class="pre">native/shuffle/src/</span></code>.</p>
<div class="pst-scrollable-table-container"><table class="table">
<thead>
<tr class="row-odd"><th class="head"><p>File</p></th>
-<th class="head"><p>Location</p></th>
<th class="head"><p>Description</p></th>
</tr>
</thead>
<tbody>
<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">shuffle_writer.rs</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">native/core/src/execution/shuffle/</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">ShuffleWriterExec</span></code> plan and partitioners. Main shuffle logic.</p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">ShuffleWriterExec</span></code>, the DataFusion <code class="docutils literal notranslate"><span class="pre">ExecutionPlan</span></code> that drives partitioning and writing.</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">comet_partitioning.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">CometPartitioning</span></code> enum defining the partition schemes (Hash, Range, Single, RoundRobin).</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">partitioners/multi_partition.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">MultiPartitionShuffleRepartitioner</span></code> for hash, range, and round robin partitioning.</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">partitioners/single_partition.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">SinglePartitionShufflePartitioner</span></code> for the single partition case.</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">writers/shuffle_block_writer.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">ShuffleBlockWriter</span></code> and the <code class="docutils literal notranslate"><span class="pre">CompressionCodec</span></code> enum. Arrow IPC encoding with compression.</p></td>
</tr>
-<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">codec.rs</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">native/core/src/execution/shuffle/</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">ShuffleBlockWriter</span></code> for Arrow IPC encoding with compression. Also handles decoding.</p></td>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">writers/partition_writer.rs</span></code></p></td>
+<td><p>The <code class="docutils literal notranslate"><span class="pre">PartitionWriter</span></code> trait. Implemented by <code class="docutils literal notranslate"><span class="pre">LocalPartitionWriter</span></code> (<code class="docutils literal notranslate"><span class="pre">writers/local/</span></code>) and <code class="docutils literal notranslate"><span class="pre">RssPartitionWriter</span></code> (<code class="docutils literal notranslate"><span class="pre">writers/rss/</span>< [...]
</tr>
-<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">comet_partitioning.rs</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">native/core/src/execution/shuffle/</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">CometPartitioning</span></code> enum defining partition schemes (Hash, Range, Single).</p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">writers/buf_batch_writer.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">BufBatchWriter</span></code>, which coalesces sub-<code class="docutils literal notranslate"><span class="pre">batch_size</span></code> batches through Arrow’s <code class="docutils literal notranslate"><span class="pre">BatchCoalescer</span></code> before serializing.</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">writers/local/spill.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">PartitionedSpill</span></code>, the one spill file shared by every output partition, and its per-partition byte ranges.</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">ipc.rs</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">read_ipc_compressed</span></code> and <code class="docutils literal notranslate"><span class="pre">read_ipc_compressed_validated</span></code>, the decode side used by the shuffle reader.</p></td>
</tr>
</tbody>
</table>
@@ -737,6 +767,17 @@ Spark’s <code class="docutils literal notranslate"><span class="pre">IndexShuf
</section>
<section id="read-path">
<h3>Read Path<a class="headerlink" href="#read-path" title="Link to this heading">#</a></h3>
+<p>There are two read paths, and which one runs is decided at plan serialization time rather than at
+read time. See <a class="reference internal" href="#direct-read-shufflescan">Direct Read</a> below for how the choice is made.</p>
+<p><strong>Direct read</strong>, the default, decodes inside the native plan:</p>
+<ol class="arabic simple">
+<li><p><code class="docutils literal notranslate"><span class="pre">CometBlockStoreShuffleReader.readAsRawStream()</span></code> concatenates the fetched block streams and
+skips decoding entirely.</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator</span></code> hands native one compressed block at a time.</p></li>
+<li><p>The native <code class="docutils literal notranslate"><span class="pre">ShuffleScanExec</span></code> decodes each block with <code class="docutils literal notranslate"><span class="pre">read_ipc_compressed</span></code> and feeds the
+resulting <code class="docutils literal notranslate"><span class="pre">RecordBatch</span></code> straight into the plan. No Arrow FFI export or import happens.</p></li>
+</ol>
+<p><strong>JVM decode</strong>, used when direct read does not apply:</p>
<ol class="arabic simple">
<li><p><code class="docutils literal notranslate"><span class="pre">CometBlockStoreShuffleReader</span></code> fetches shuffle blocks via <code class="docutils literal notranslate"><span class="pre">ShuffleBlockFetcherIterator</span></code>.</p></li>
<li><p>For each block, <code class="docutils literal notranslate"><span class="pre">NativeBatchDecoderIterator</span></code>:</p>
@@ -752,6 +793,67 @@ Spark’s <code class="docutils literal notranslate"><span class="pre">IndexShuf
</ol>
</section>
</section>
+<section id="direct-read-shufflescan">
+<h2>Direct Read (ShuffleScan)<a class="headerlink" href="#direct-read-shufflescan" title="Link to this heading">#</a></h2>
+<p>Direct read lets a native operator consume shuffle output without the batch ever being decoded in
+the JVM or crossing Arrow FFI. It is controlled by <code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.directRead.enabled</span></code>, which
+defaults to <code class="docutils literal notranslate"><span class="pre">true</span></code> and requires <code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.enabled</span></code>. It applies to both native shuffle and
+JVM columnar shuffle, because both write the same Arrow IPC block format.</p>
+<section id="how-the-path-is-selected">
+<h3>How the path is selected<a class="headerlink" href="#how-the-path-is-selected" title="Link to this heading">#</a></h3>
+<p><code class="docutils literal notranslate"><span class="pre">CometExchangeSink.shouldUseShuffleScan</span></code> (<code class="docutils literal notranslate"><span class="pre">spark/src/main/scala/org/apache/comet/serde/operator/CometSink.scala</span></code>)
+decides during plan serialization. When direct read is enabled and the sink’s input is a Comet shuffle
+exchange, <code class="docutils literal notranslate"><span class="pre">convertToShuffleScan</span></code> emits a <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code> operator. When either is false the sink falls
+through to the base <code class="docutils literal notranslate"><span class="pre">CometSink.convert</span></code>, which emits the usual <code class="docutils literal notranslate"><span class="pre">Scan</span></code>.</p>
+<p>The two are not alternatives on failure. If any output type fails <code class="docutils literal notranslate"><span class="pre">supportedSinkDataType</span></code>,
+<code class="docutils literal notranslate"><span class="pre">convertToShuffleScan</span></code> records the fallback reason <code class="docutils literal notranslate"><span class="pre">Unsupported</span> <span class="pre">data</span> <span class="pre">type</span> <span class="pre">for</span> <span class="pre">shuffle</span> <span class="pre">direct</span> <span class="pre">read</span></code>
+and returns <code class="docutils literal notranslate"><span class="pre">None</span></code>. It does not retry as a regular <code class="docutils literal notranslate"><span class="pre">Scan</span></code>, and retrying would not help, because
+<code class="docutils literal notranslate"><span class="pre">CometSink.convert</span></code> gates on the same <code class="docutils literal notranslate"><span class="pre">supportedSinkDataType</span></code>. <code class="docutils literal notranslate"><span class="pre">CometExecRule</span></code> calls this as
+<code class="docutils literal notranslate"><span class="pre">convertToComet(s,</span> <span class="pre">CometExchangeSink).getOrElse(s)</span></code>, so <code class="docutils literal notranslate"><span class="pre">None</span></code> leaves the original Spark shuffle stage
+in the plan. An unsupported output type therefore means the stage is not converted natively at all,
+not that it is served over the FFI read path instead.</p>
+<p><strong>The protobuf is the source of truth for which slots are direct read, not the config.</strong>
+<code class="docutils literal notranslate"><span class="pre">findShuffleScanIndices</span></code> (<code class="docutils literal notranslate"><span class="pre">operators.scala</span></code>) walks the serialized plan, counting scan slots in order
+and collecting the indices that carry a <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code>. JVM-side input dispatch reads that set rather
+than re-checking the config, so the two cannot disagree. Anything changing the serde condition has to
+leave that walk consistent with it.</p>
+</section>
+<section id="how-blocks-reach-native">
+<h3>How blocks reach native<a class="headerlink" href="#how-blocks-reach-native" title="Link to this heading">#</a></h3>
+<p><code class="docutils literal notranslate"><span class="pre">CometExecRDD.resolveInputObjects</span></code> fills one input slot per scan input, in scan-input order:</p>
+<ul class="simple">
+<li><p>A slot in <code class="docutils literal notranslate"><span class="pre">shuffleScanIndices</span></code> gets a <code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator</span></code>, obtained from
+<code class="docutils literal notranslate"><span class="pre">CometShuffledBatchRDD.computeAsShuffleBlockIterator</span></code>. A slot marked as a shuffle scan whose RDD is
+not a <code class="docutils literal notranslate"><span class="pre">CometShuffledBatchRDD</span></code> throws <code class="docutils literal notranslate"><span class="pre">CometRuntimeException</span></code>.</p></li>
+<li><p>Every other slot gets the <code class="docutils literal notranslate"><span class="pre">ArrowArrayStream</span></code> exported by the ordinary FFI path.</p></li>
+</ul>
+<p><code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator</span></code> reads a 16-byte header per block: an 8-byte compressed length, which
+includes the field count but not itself, followed by an 8-byte field count that is discarded because
+the schema comes from the <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code> protobuf fields. The compressed body is read into a reused
+<code class="docutils literal notranslate"><span class="pre">DirectByteBuffer</span></code>, <strong>valid only until the next <code class="docutils literal notranslate"><span class="pre">hasNext()</span></code> call</strong>. Native must fully consume it
+before pulling the next block, which <code class="docutils literal notranslate"><span class="pre">read_ipc_compressed</span></code> satisfies because it allocates fresh
+native memory for the decoded data.</p>
+</section>
+<section id="native-side">
+<h3>Native side<a class="headerlink" href="#native-side" title="Link to this heading">#</a></h3>
+<p><code class="docutils literal notranslate"><span class="pre">ShuffleScanExec</span></code> (<code class="docutils literal notranslate"><span class="pre">native/core/src/execution/operators/shuffle_scan.rs</span></code>) pulls blocks through the
+iterator’s <code class="docutils literal notranslate"><span class="pre">hasNext()</span></code> and <code class="docutils literal notranslate"><span class="pre">getBuffer()</span></code> JNI methods and decodes them with <code class="docutils literal notranslate"><span class="pre">read_ipc_compressed</span></code>.
+Two details matter when changing it:</p>
+<ul class="simple">
+<li><p><code class="docutils literal notranslate"><span class="pre">get_next_batch</span></code> is called from outside <code class="docutils literal notranslate"><span class="pre">poll_next</span></code>, because JNI calls cannot be made from tokio
+worker threads. A change that moves the JNI call into the stream’s <code class="docutils literal notranslate"><span class="pre">poll_next</span></code> breaks this.</p></li>
+<li><p>Dictionary-encoded columns are unpacked to their value type by <code class="docutils literal notranslate"><span class="pre">unpack_dictionary</span></code>, so the schema
+the plan sees matches the declared <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code> fields.</p></li>
+</ul>
+<p>Blocks are validated before decoding when <code class="docutils literal notranslate"><span class="pre">CometShuffleBlockIterator.requiresValidation()</span></code> is true,
+which happens when the underlying stream implements <code class="docutils literal notranslate"><span class="pre">CometShuffleReadFailureHandler</span></code> so a corrupt
+block can be reported back as a shuffle read failure rather than a native decode panic.</p>
+<p>Celeborn reuses the same expected-schema contract from the JVM side. <code class="docutils literal notranslate"><span class="pre">CometCelebornShuffleReader</span></code>
+builds a <code class="docutils literal notranslate"><span class="pre">ShuffleScan</span></code> message from the dependency’s output attributes and passes it to
+<code class="docutils literal notranslate"><span class="pre">NativeBatchDecoderIterator</span></code>, so remote logical types are validated before Arrow import and an
+unsupported type fails before any shuffle resource is acquired.</p>
+</section>
+</section>
<section id="partitioning">
<h2>Partitioning<a class="headerlink" href="#partitioning" title="Link to this heading">#</a></h2>
<section id="hash-partitioning">
@@ -792,6 +894,10 @@ However, unlike true round robin which cycles through partitions row-by-row, has
assignment only provides even distribution when the data has sufficient variation in the
hashed columns. Data with low cardinality or identical values may result in skewed partition
sizes.</p>
+<p>Because Spark assigns round robin partitions by sorting rows on their binary <code class="docutils literal notranslate"><span class="pre">UnsafeRow</span></code> form,
+which Arrow’s layout does not reproduce, unsorted output can land in different partitions than
+Spark’s. Sorted output is identical. That difference is why
+<code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.roundrobin.enabled</span></code> defaults to <code class="docutils literal notranslate"><span class="pre">false</span></code>.</p>
</section>
</section>
<section id="memory-management">
@@ -802,16 +908,29 @@ sizes.</p>
<li><p><strong>Spill triggers</strong>: Partitions spill to disk when the memory pool denies an allocation, or
when the buffered bytes reach <code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.maxBufferBytes</span></code>. That config defaults to
0, which disables the fixed limit and leaves memory pressure as the only trigger.</p></li>
-<li><p><strong>Per-partition spilling</strong>: Each partition has its own spill file. Multiple spills for a
-partition are concatenated when writing the final output.</p></li>
+<li><p><strong>One spill file per task</strong>: Every output partition spills into the same file, not one file per
+partition. <code class="docutils literal notranslate"><span class="pre">PartitionedSpill</span></code> records the byte ranges each partition’s blocks occupy, in write
+order, and the final output copies a partition’s ranges back in that order.</p></li>
<li><p><strong>Scratch space</strong>: Reusable buffers for partition ID computation to reduce allocations.</p></li>
</ul>
-<p>The <code class="docutils literal notranslate"><span class="pre">MultiPartitionShuffleRepartitioner</span></code> manages:</p>
+<p>The <code class="docutils literal notranslate"><span class="pre">MultiPartitionShuffleRepartitioner</span></code> holds:</p>
<ul class="simple">
-<li><p><code class="docutils literal notranslate"><span class="pre">PartitionBuffer</span></code>: In-memory buffer for each partition</p></li>
-<li><p><code class="docutils literal notranslate"><span class="pre">SpillFile</span></code>: Temporary file for spilled data</p></li>
-<li><p>Memory tracking via <code class="docutils literal notranslate"><span class="pre">MemoryConsumer</span></code> trait</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">buffered_batches</span></code>, a <code class="docutils literal notranslate"><span class="pre">Vec<RecordBatch></span></code> of incoming batches, alongside <code class="docutils literal notranslate"><span class="pre">partition_indices</span></code>
+recording which rows of those batches belong to each partition. Rows are not copied into
+per-partition buffers as they arrive.</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">reservation</span></code>, a <code class="docutils literal notranslate"><span class="pre">MemoryReservation</span></code> charged for the bytes each buffered batch newly pins.
+<code class="docutils literal notranslate"><span class="pre">pinned_buffers</span></code> tracks backing buffer start addresses so one allocation shared by many sliced
+batches is charged once rather than once per slice. Charging a per-batch size instead would
+overstate memory by the slice count and spill spuriously.</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">max_buffer_bytes</span></code>, the optional fixed spill threshold described above. <code class="docutils literal notranslate"><span class="pre">None</span></code> leaves pool
+pressure as the only trigger.</p></li>
+<li><p><code class="docutils literal notranslate"><span class="pre">scratch</span></code>, reusable buffers for partition ID computation.</p></li>
</ul>
+<p>The spill file is owned by <code class="docutils literal notranslate"><span class="pre">PartitionedSpill</span></code> in <code class="docutils literal notranslate"><span class="pre">writers/local/spill.rs</span></code>. It holds one DataFusion
+<code class="docutils literal notranslate"><span class="pre">SpillFile</span></code>, created lazily on the first spill and shared by every output partition, and tracks per
+partition the byte ranges holding that partition’s blocks in write order. A partial write sets a
+<code class="docutils literal notranslate"><span class="pre">failed</span></code> flag, because a write that stops midway leaves the recorded ranges unable to describe the
+file. The single-partition writer writes straight to the output file and never spills.</p>
</section>
<section id="compression">
<h2>Compression<a class="headerlink" href="#compression" title="Link to this heading">#</a></h2>
@@ -860,19 +979,47 @@ independently compressed, allowing parallel decompression during reads.</p>
<td><p><code class="docutils literal notranslate"><span class="pre">auto</span></code></p></td>
<td><p>Shuffle mode: <code class="docutils literal notranslate"><span class="pre">native</span></code>, <code class="docutils literal notranslate"><span class="pre">jvm</span></code>, or <code class="docutils literal notranslate"><span class="pre">auto</span></code></p></td>
</tr>
-<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.compression.codec</span></code></p></td>
-<td><p><code class="docutils literal notranslate"><span class="pre">zstd</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.directRead.enabled</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">true</span></code></p></td>
+<td><p>Decode shuffle blocks in native code, bypassing Arrow FFI</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.compression.codec</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">lz4</span></code></p></td>
<td><p>Compression codec</p></td>
</tr>
-<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.compression.zstd.level</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.compression.zstd.level</span></code></p></td>
<td><p><code class="docutils literal notranslate"><span class="pre">1</span></code></p></td>
<td><p>Zstd compression level</p></td>
</tr>
-<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.writeBufferSize</span></code></p></td>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.writeBufferSize</span></code></p></td>
<td><p><code class="docutils literal notranslate"><span class="pre">1MB</span></code></p></td>
<td><p>Write buffer size</p></td>
</tr>
-<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.jvm.batchSize</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.maxBufferBytes</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">0</span></code></p></td>
+<td><p>Fixed spill threshold. <code class="docutils literal notranslate"><span class="pre">0</span></code> disables it, leaving pool pressure</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.hash.enabled</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">true</span></code></p></td>
+<td><p>Allow <code class="docutils literal notranslate"><span class="pre">HashPartitioning</span></code> on the native path</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.hash.nested.enabled</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">false</span></code></p></td>
+<td><p>Allow struct and array hash keys, and map keys on Spark 4.0+</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.range.enabled</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">true</span></code></p></td>
+<td><p>Allow <code class="docutils literal notranslate"><span class="pre">RangePartitioning</span></code> on the native path</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.roundrobin.enabled</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">false</span></code></p></td>
+<td><p>Allow <code class="docutils literal notranslate"><span class="pre">RoundRobinPartitioning</span></code> on the native path</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.native.partitioning.roundrobin.maxHashColumns</span></code></p></td>
+<td><p><code class="docutils literal notranslate"><span class="pre">0</span></code></p></td>
+<td><p>Columns to hash for round robin. <code class="docutils literal notranslate"><span class="pre">0</span></code> hashes all of them</p></td>
+</tr>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.shuffle.jvm.batchSize</span></code></p></td>
<td><p><code class="docutils literal notranslate"><span class="pre">8192</span></code></p></td>
<td><p>Target rows per batch</p></td>
</tr>
@@ -981,6 +1128,12 @@ independently compressed, allowing parallel decompression during reads.</p>
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#read-path">Read Path</a></li>
</ul>
</li>
+<li class="toc-h2 nav-item toc-entry"><a class="reference internal nav-link" href="#direct-read-shufflescan">Direct Read (ShuffleScan)</a><ul class="nav section-nav flex-column">
+<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#how-the-path-is-selected">How the path is selected</a></li>
+<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#how-blocks-reach-native">How blocks reach native</a></li>
+<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#native-side">Native side</a></li>
+</ul>
+</li>
<li class="toc-h2 nav-item toc-entry"><a class="reference internal nav-link" href="#partitioning">Partitioning</a><ul class="nav section-nav flex-column">
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#hash-partitioning">Hash Partitioning</a></li>
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#range-partitioning">Range Partitioning</a></li>
diff --git a/searchindex.js b/searchindex.js
index 4d05eae956..b11493b118 100644
--- a/searchindex.js
+++ b/searchindex.js
@@ -1 +1 @@
-Search.setIndex({"alltitles": {"!": [[56, "id1"]], "%": [[54, "id1"]], "&": [[44, "id1"]], "*": [[54, "id2"]], "+": [[54, "id3"]], "-": [[54, "id4"]], "/": [[54, "id5"]], "1. Format Your Code": [[41, "format-your-code"]], "1. Install Comet": [[62, "install-comet"], [73, "install-comet"]], "1. Native Operators (nativeExecs map)": [[28, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[41, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[62, "clone-iceberg-and-apply- [...]
\ No newline at end of file
+Search.setIndex({"alltitles": {"!": [[56, "id1"]], "%": [[54, "id1"]], "&": [[44, "id1"]], "*": [[54, "id2"]], "+": [[54, "id3"]], "-": [[54, "id4"]], "/": [[54, "id5"]], "1. Format Your Code": [[41, "format-your-code"]], "1. Install Comet": [[62, "install-comet"], [73, "install-comet"]], "1. Native Operators (nativeExecs map)": [[28, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[41, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[62, "clone-iceberg-and-apply- [...]
\ No newline at end of file
diff --git a/user-guide/latest/configs.html b/user-guide/latest/configs.html
index 5ad16a0e2e..5caa8547fd 100644
--- a/user-guide/latest/configs.html
+++ b/user-guide/latest/configs.html
@@ -933,11 +933,15 @@ under the License.
<td><p>Whether to allow incompatibility for operator: DataWritingCommandExec. False by default. Can be overridden with SPARK_COMET_OPERATOR_DATAWRITINGCOMMANDEXEC_ALLOWINCOMPATIBLE env variable It can be overridden by the environment variable <code class="docutils literal notranslate"><span class="pre">SPARK_COMET_OPERATOR_DATAWRITINGCOMMANDEXEC_ALLOWINCOMPATIBLE</span></code>.</p></td>
<td><p>false</p></td>
</tr>
-<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.enabled</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.operator.WriteFilesExec.allowIncompatible</span></code></p></td>
+<td><p>Whether to allow incompatibility for operator: WriteFilesExec. False by default. Can be overridden with SPARK_COMET_OPERATOR_WRITEFILESEXEC_ALLOWINCOMPATIBLE env variable It can be overridden by the environment variable <code class="docutils literal notranslate"><span class="pre">SPARK_COMET_OPERATOR_WRITEFILESEXEC_ALLOWINCOMPATIBLE</span></code>.</p></td>
+<td><p>false</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.enabled</span></code></p></td>
<td><p>Whether to enable Spark to Arrow columnar conversion. When this is turned on, Comet will convert operators in <code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.supportedOperatorList</span></code> into Arrow columnar format before processing.</p></td>
<td><p>false</p></td>
</tr>
-<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.supportedOperatorList</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.supportedOperatorList</span></code></p></td>
<td><p>A comma-separated list of operators that will be converted to Arrow columnar format when <code class="docutils literal notranslate"><span class="pre">spark.comet.sparkToColumnar.enabled</span></code> is true.</p></td>
<td><p>Range,InMemoryTableScan,RDDScan,OneRowRelation</p></td>
</tr>
diff --git a/user-guide/latest/installation.html b/user-guide/latest/installation.html
index 8afe68b0fd..f36d55f8ad 100644
--- a/user-guide/latest/installation.html
+++ b/user-guide/latest/installation.html
@@ -954,14 +954,17 @@ source releases, or from the latest code in the GitHub repository.</p>
<div class="highlight-scala notranslate"><div class="highlight"><pre><span></span><span class="n">scala</span><span class="o">></span><span class="w"> </span><span class="p">(</span><span class="mi">0</span><span class="w"> </span><span class="n">until</span><span class="w"> </span><span class="mi">10</span><span class="p">).</span><span class="n">toDF</span><span class="p">(</span><span class="s">"a"</span><span class="p">).</span><span class="n">write</span><span class="p">.</span>< [...]
</pre></div>
</div>
-<p>Comet will log output similar to:</p>
+<p>Comet will log output similar to this on Spark 4.0 and later:</p>
<div class="highlight-shell notranslate"><div class="highlight"><pre><span></span>INFO<span class="w"> </span>core/src/lib.rs:<span class="w"> </span>Comet<span class="w"> </span>native<span class="w"> </span>library<span class="w"> </span>version<span class="w"> </span><span class="m">1</span>.1.0-SNAPSHOT<span class="w"> </span>initialized
WARN<span class="w"> </span>CometExecRule:<span class="w"> </span>Comet<span class="w"> </span>cannot<span class="w"> </span>execute<span class="w"> </span>some<span class="w"> </span>parts<span class="w"> </span>of<span class="w"> </span>this<span class="w"> </span>plan<span class="w"> </span>natively<span class="w"> </span><span class="o">(</span><span class="nb">set</span><span class="w"> </span>spark.comet.explain.fallback.enabled<span class="o">=</span><span class="nb">false</span>< [...]
-<span class="w"> </span>Execute<span class="w"> </span>InsertIntoHadoopFsRelationCommand<span class="w"> </span><span class="o">[</span>COMET:<span class="w"> </span>Native<span class="w"> </span>support<span class="w"> </span><span class="k">for</span><span class="w"> </span>operator<span class="w"> </span>DataWritingCommandExec<span class="w"> </span>is<span class="w"> </span>disabled.<span class="w"> </span>Set<span class="w"> </span>spark.comet.parquet.write.enabled<span class="o">= [...]
-+-<span class="w"> </span>WriteFiles
+<span class="w"> </span>Execute<span class="w"> </span>InsertIntoHadoopFsRelationCommand
++-<span class="w"> </span>WriteFiles<span class="w"> </span><span class="o">[</span>COMET:<span class="w"> </span>Native<span class="w"> </span>support<span class="w"> </span><span class="k">for</span><span class="w"> </span>operator<span class="w"> </span>WriteFilesExec<span class="w"> </span>is<span class="w"> </span>disabled.<span class="w"> </span>Set<span class="w"> </span>spark.comet.parquet.write.enabled<span class="o">=</span><span class="nb">true</span><span class="w"> </span>to [...]
<span class="w"> </span>+-<span class="w"> </span>LocalTableScan<span class="w"> </span><span class="o">[</span>COMET:<span class="w"> </span>Native<span class="w"> </span>support<span class="w"> </span><span class="k">for</span><span class="w"> </span>operator<span class="w"> </span>LocalTableScanExec<span class="w"> </span>is<span class="w"> </span>disabled.<span class="w"> </span>Set<span class="w"> </span>spark.comet.exec.localTableScan.enabled<span class="o">=</span><span class=" [...]
</pre></div>
</div>
+<p>On Spark 3.4 and 3.5 the native writer replaces the whole write command rather than just the
+per-task write, so the same message appears on <code class="docutils literal notranslate"><span class="pre">Execute</span> <span class="pre">InsertIntoHadoopFsRelationCommand</span></code> and
+names <code class="docutils literal notranslate"><span class="pre">DataWritingCommandExec</span></code>.</p>
<p>Query the data from the test source and check:</p>
<ul class="simple">
<li><p>INFO message shows the native Comet library has been initialized.</p></li>
diff --git a/user-guide/latest/operators.html b/user-guide/latest/operators.html
index 1a64b8e313..f9f705fbed 100644
--- a/user-guide/latest/operators.html
+++ b/user-guide/latest/operators.html
@@ -1051,9 +1051,13 @@ omitted from the tables below and may be reconsidered based on demand:</p>
</tr>
</thead>
<tbody>
-<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">DataWritingCommandExec</span></code></p></td>
+<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">WriteFilesExec</span></code></p></td>
<td><p>⚠️</p></td>
-<td><p>Experimental native Parquet writes, disabled by default (opt-in).</p></td>
+<td><p>Spark 4.0+. Experimental native Parquet writes, disabled by default (opt-in). Non-partitioned, non-bucketed writes only, and not when <code class="docutils literal notranslate"><span class="pre">spark.sql.files.maxRecordsPerFile</span></code> is set.</p></td>
+</tr>
+<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">DataWritingCommandExec</span></code></p></td>
+<td><p>⚠️</p></td>
+<td><p>Spark 3.4/3.5 only. Experimental native Parquet writes, disabled by default (opt-in). Replaced by <code class="docutils literal notranslate"><span class="pre">WriteFilesExec</span></code> on Spark 4.0+ and removed with Spark 3.x support.</p></td>
</tr>
</tbody>
</table>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org