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/12 20:42:57 UTC
(datafusion-comet) branch asf-site updated: Publish built docs triggered by 23a9a7cfb9e0775b4f3876c19dd2e80b4242aa0d
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
The following commit(s) were added to refs/heads/asf-site by this push:
new 0f734a0a56 Publish built docs triggered by 23a9a7cfb9e0775b4f3876c19dd2e80b4242aa0d
0f734a0a56 is described below
commit 0f734a0a56dec8e17503437665adec8826068779
Author: github-actions[bot] <gi...@users.noreply.github.com>
AuthorDate: Mon Oct 12 20:42:52 2026 +0000
Publish built docs triggered by 23a9a7cfb9e0775b4f3876c19dd2e80b4242aa0d
---
...id-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg | 2 +-
_sources/contributor-guide/iceberg-writes.md.txt | 44 ++++++++++++++--------
.../contributor-guide/memory_management.md.txt | 18 ++++++---
_sources/user-guide/latest/iceberg-writes.md.txt | 37 ++++++++++++------
contributor-guide/iceberg-writes.html | 26 ++++++++++---
contributor-guide/memory_management.html | 18 ++++++---
searchindex.js | 2 +-
user-guide/latest/iceberg-writes.html | 36 ++++++++++++------
8 files changed, 126 insertions(+), 57 deletions(-)
diff --git a/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg b/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
index 571f46438b..34c5717fcb 100644
--- a/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
+++ b/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
@@ -1 +1 @@
-<svg id="my-svg" width="100%" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" class="flowchart" style="max-width: 511.73px; background-color: transparent;" viewBox="0 0 511.7299499511719 966" role="graphics-document document" aria-roledescription="flowchart-v2"><style>#my-svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#my-svg [...]
\ No newline at end of file
+<svg id="my-svg" width="100%" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" class="flowchart" style="max-width: 511.73px; background-color: transparent;" viewBox="0 0 511.7299499511719 966" role="graphics-document document" aria-roledescription="flowchart-v2"><style>#my-svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#my-svg [...]
\ No newline at end of file
diff --git a/_sources/contributor-guide/iceberg-writes.md.txt b/_sources/contributor-guide/iceberg-writes.md.txt
index 2cdfb9038b..6f930733b1 100644
--- a/_sources/contributor-guide/iceberg-writes.md.txt
+++ b/_sources/contributor-guide/iceberg-writes.md.txt
@@ -223,7 +223,7 @@ per task, and `PartitionWriterBuilder` builds the rest of the stack once per par
partition's Parquet writer properties:
```text
-UnpartitionedWriter | FanoutWriter | ClusteredWriter
+UnpartitionedWriter | FanoutPartitions | ClusteredWriter
-> PartitionWriterBuilder, per partition:
ParquetWriterBuilder -> RollingFileWriterBuilder -> DataFileWriterBuilder
```
@@ -270,8 +270,9 @@ Points where Comet adapts iceberg-rust to match iceberg-java:
- **Field ids and casting.** `decorate_batch_with_field_ids` casts each batch to the
field-id-annotated Arrow schema derived from the Iceberg schema, with `safe: false`, so a type
mismatch fails the task instead of writing NULLs.
-- **Deterministic output order.** iceberg-rust's `FanoutWriter` closes its writers out of a
- `HashMap`, so the fanout path sorts its `DataFile`s by path before returning them
+- **Deterministic output order.** `FanoutPartitions`, Comet's version of iceberg-rust's
+ `FanoutWriter`, closes its files in an order that depends on how the rows arrived and on which
+ partitions it closed early, so the fanout path sorts its `DataFile`s by path before returning them
([#5776](https://github.com/apache/datafusion-comet/issues/5776)). Manifest order becomes the row
order of an unordered read, so any map iteration that reaches the output needs the same care.
@@ -281,8 +282,21 @@ hold in memory. It hands `ParquetWriter` each file's `OutputFile` behind a `Coun
writer counts the bytes that leave memory on their way to storage, and a file reports what it has
written less those. When they leave depends on the storage (`StorageWrites`). After every batch
`run_write_task` resizes the task's reservation to what the open files report plus the rows each
-`PartitionFeed` holds, for the dictionary choice or for pacing, and a resize the pool refuses fails
-the task. What that figure covers, and what it misses, is described under
+`PartitionFeed` holds, for the dictionary choice or for pacing (`InnerWriter::reserve`). When the
+pool refuses a fanout write, `reserve` writes out and closes the partitions holding the most, in
+their open file and their feed, until the resize succeeds, and `run_write_task` counts them in the
+`files_closed_early` metric. Taking the largest first frees the most memory per file closed, so
+each refusal closes as few files as it can. Each partition's files report to a child of the task's
+`OpenFileMemory`, which is how `reserve` finds them. That is why the fanout path uses
+`FanoutPartitions` rather than iceberg-rust's `FanoutWriter`, which cannot close one partition's
+writer. A closed partition's next rows open a new file with the properties its first file used,
+which the partition keeps. A refusal from the pool cannot fail a fanout write: everything the write
+reserves belongs to one of its partitions, so once all of them are closed it reserves nothing, and a
+resize down always succeeds. Only a refusal (`ResourcesExhausted`) makes `reserve` close
+partitions. Any other error from a resize fails the task, such as an exception Spark threw because
+a spill failed or because the task was killed while it waited for memory. An unpartitioned or
+clustered write has no file it can close early, so a resize the pool refuses fails its task. What
+the reserved figure covers, and what it misses, is described under
[Native writers](memory_management.md#native-writers).
`FileIO` comes from `load_file_io` in `iceberg_common.rs`, shared with the native scan. It picks
@@ -416,16 +430,16 @@ the writer: run the write suites and the Iceberg Spark tests. The pin policy is
## Testing
-| Suite | What it covers |
-| ---------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `CometIcebergWriteActionSuite` | End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, a fanout write outgrowing the memory pool, AQE re-planning. |
-| `CometIcebergWriteDetectionSuite` | One case per eligibility rule, accepted and declined. |
-| `CometIcebergSystemFunctionSuite` | Native `bucket`, `truncate`, `years`/`months`/`days`/`hours`, which keep a partitioned write's hash distribution and sort native end to end. |
-| `CometIcebergRewriteActionSuite` | Iceberg's `rewrite_data_files` with the split plan and the native writer. |
-| `IcebergWriteProtoTranslationSuite` | Translation of properties into `IcebergParquetWriteSettings` and the writer mode. |
-| Rust tests in `iceberg_write.rs` and in `iceberg_partition_*.rs` | File rolling on the 1000-row grid, fanout order, clustered input checks, cleanup guard, manifest round trip, memory reservation, partition path rendering, partition values past `chrono`'s calendar. |
-| Rust tests in `iceberg_dictionary.rs` | The per-column dictionary choice against answers recorded from parquet-mr, including columns either side of its cut-off. |
-| `CometIcebergWriteBenchmark` | Native versus iceberg-java for unpartitioned, clustered, fanout and copy-on-write delete writes. It checks each arm's plan before timing it. |
+| Suite | What it covers |
+| ---------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| `CometIcebergWriteActionSuite` | End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, writes outgrowing the memory pool, AQE re-planning. |
+| `CometIcebergWriteDetectionSuite` | One case per eligibility rule, accepted and declined. |
+| `CometIcebergSystemFunctionSuite` | Native `bucket`, `truncate`, `years`/`months`/`days`/`hours`, which keep a partitioned write's hash distribution and sort native end to end. |
+| `CometIcebergRewriteActionSuite` | Iceberg's `rewrite_data_files` with the split plan and the native writer. |
+| `IcebergWriteProtoTranslationSuite` | Translation of properties into `IcebergParquetWriteSettings` and the writer mode. |
+| Rust tests in `iceberg_write.rs` and in `iceberg_partition_*.rs` | File rolling on the 1000-row grid, fanout order, clustered input checks, cleanup guard, manifest round trip, memory reservation, partition path rendering, partition values past `chrono`'s calendar. |
+| Rust tests in `iceberg_dictionary.rs` | The per-column dictionary choice against answers recorded from parquet-mr, including columns either side of its cut-off. |
+| `CometIcebergWriteBenchmark` | Native versus iceberg-java for unpartitioned, clustered, fanout and copy-on-write delete writes. It checks each arm's plan before timing it. |
The Comet suites run against the Iceberg version each Spark profile pins in `spark/pom.xml`: 1.5.2
for Spark 3.4, 1.8.1 for 3.5, 1.10.0 for 4.0 and 4.2, and 1.11.0 for 4.1. Only the default profile
diff --git a/_sources/contributor-guide/memory_management.md.txt b/_sources/contributor-guide/memory_management.md.txt
index af84e4a278..9c31d3fecb 100644
--- a/_sources/contributor-guide/memory_management.md.txt
+++ b/_sources/contributor-guide/memory_management.md.txt
@@ -396,8 +396,11 @@ An operator that never calls `try_grow` is invisible to the pool no matter how m
### Native writers
Both native writers reserve what they hold between batches through a single consumer per task,
-`ParquetWriterExec[N]` or `IcebergWriteExec[N]`, resized after every batch. Neither can spill, so
-when the pool refuses a resize the task fails with a `CometNativeException` whose message starts
+`ParquetWriterExec[N]` or `IcebergWriteExec[N]`, resized after every batch. Neither can spill. A
+fanout Iceberg write gives memory back instead, by closing partitions early when the pool refuses a
+resize, as described below, so a refusal never fails it. An exception Spark throws during a resize,
+such as a failed spill or a killed task, still fails the task. When the pool refuses any other
+write's resize, the task fails with a `CometNativeException` whose message starts
`Additional allocation failed for` and names the consumer. That is a task failure Spark can retry.
Unreserved, the same memory would count only toward the container limit, where exceeding it kills
the executor.
@@ -418,10 +421,13 @@ the executor.
it has flushed since its last part was uploaded. At the default row-group size
(`write.parquet.row-group-size-bytes`, 128 MiB) that is the last row group it flushed. The
reservation also covers the rows each partition holds back, first for its dictionary choice and
- then until they fill the 1000-row unit the rolling writer is fed in. It does not cover what
- parquet-rs holds beyond the encoded size: dictionary hash tables, unencoded dictionary indices
- and buffer capacity. Native writes decline Bloom filters today, and neither figure would include
- them.
+ then until they fill the 1000-row unit the rolling writer is fed in. When the pool refuses a
+ fanout write's resize, the writer writes out and closes the partitions holding the most, file
+ and held rows together, until the resize succeeds, so the write ends with more files rather than
+ failing. Each file also reports to its partition's total, which is how the writer finds them.
+ The reservation does not cover what parquet-rs holds beyond the encoded size: dictionary hash
+ tables, unencoded dictionary indices and buffer capacity. Native writes decline Bloom filters
+ today, and neither figure would include them.
The writers register one consumer per task rather than one per open file because every consumer
registered with `fair_unified` lowers the share of every other consumer in the task. A consumer per
diff --git a/_sources/user-guide/latest/iceberg-writes.md.txt b/_sources/user-guide/latest/iceberg-writes.md.txt
index 2d84caeaca..cea6ca42a9 100644
--- a/_sources/user-guide/latest/iceberg-writes.md.txt
+++ b/_sources/user-guide/latest/iceberg-writes.md.txt
@@ -315,13 +315,24 @@ The native writer's buffers are charged to Comet's memory pool, the off-heap bud
native operators draw on, where iceberg-java's buffers sit on the JVM heap. A fanout write keeps a
data file open for every partition a task writes to. Each open file holds the row group it is
writing in memory, up to `write.parquet.row-group-size-bytes`, and on S3 or GCS also the last row
-group it flushed, which is uploaded once the next one is complete or the file closes. So a task
-writing to many partitions needs memory in proportion to them. When the pool cannot grant it, the
-task fails with a `CometNativeException` reading `Additional allocation failed for IcebergWriteExec`
-instead of exceeding the executor's memory, and Spark retries it like any other task failure. Such a
-write fits in less memory with the fanout writer disabled (`write.spark.fanout.enabled=false`):
-Spark then sorts each task's rows by partition, and the task keeps one file open at a time. A
-smaller row-group size also helps. Otherwise the write needs a larger `spark.memory.offHeap.size`.
+group it flushed, which is uploaded once the next one is complete or the file closes. Each
+partition also holds the rows that have not reached its file yet. So a task writing to many
+partitions needs memory in proportion to them. When the pool cannot grant it, the task writes out
+and closes the partitions holding the most memory until what is left fits, and a closed
+partition's next rows open a new file. The write then finishes with more, smaller files than
+iceberg-java's would. When every partition keeps getting rows, a task given an eighth of the memory
+its partitions need writes about four files per partition instead of one, however many partitions
+it writes to. The `files closed early to free memory` metric of its `CometIcebergWrite` operator
+counts the files it closed early. Disabling the fanout writer (`write.spark.fanout.enabled=false`)
+avoids the extra files: Spark then sorts each task's rows by partition, and the task keeps one file
+open at a time. A larger `spark.memory.offHeap.size` also helps, and Iceberg's `rewrite_data_files`
+procedure compacts small files once they are written.
+
+A write that keeps one file open, unpartitioned or clustered, has no partition to close. When that
+file outgrows the pool, the task fails with a `CometNativeException` reading
+`Additional allocation failed for IcebergWriteExec` instead of exceeding the executor's memory, and
+Spark retries it like any other task failure. A smaller `write.parquet.row-group-size-bytes` or a
+larger `spark.memory.offHeap.size` lets such a write fit.
Partial results are never committed. The commit set is exactly the commit messages returned by
successful tasks — a failed task contributes none — and if the job fails, the driver-side
@@ -416,6 +427,11 @@ a data file but not what any reader computes from it:
they may cross the target several grid steps apart, and the resulting files can differ in row
count by an arbitrary number of 1000-row blocks. Do not rely on file-layout parity between the
two writers; rely only on each file rolling on its own 1000-row boundary.
+- A fanout write that the memory pool cannot hold closes partitions before the task ends (see
+ [Failure handling](#failure-handling)), where iceberg-java's fanout writer keeps every file open
+ until then. So a partition can have more, smaller files than iceberg-java writes, and a file
+ closed early ends off the 1000-row grid, as the last file of a task does. A partition closed
+ before its rows filled a first page makes its dictionary choice from the rows it has.
- A fanout write lists a task's data files in file-path order, where iceberg-java lists them in
its own `StructLikeMap` iteration order. Both are stable across runs, and neither is a
documented ordering, but the manifest entry order becomes the scan-task order and so the row
@@ -423,10 +439,9 @@ a data file but not what any reader computes from it:
commit gives each file's rows, so the same rows can get different `_row_id` values from the two
writers. The ids are unique either way, and across tasks iceberg-java's own assignment already
depends on the order in which the tasks finish. Only the sorted order is reproducible on the
- native path: iceberg-rust's `FanoutWriter` closes its per-partition writers out of a `HashMap`,
- which under Rust's per-process `RandomState` would otherwise give a different order on every
- run. Clustered and unpartitioned writes append in creation order on both paths and are
- unaffected.
+ native path: the order the native writer closes its files in depends on the order rows arrive in
+ and on which partitions it closes early to free memory. Clustered and unpartitioned writes append
+ in creation order on both paths and are unaffected.
- Compressed page bytes are implementation-defined: the codec and any explicit level are
translated, but parquet-rs and parquet-mr embed different encoder implementations and
defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even
diff --git a/contributor-guide/iceberg-writes.html b/contributor-guide/iceberg-writes.html
index e964a697f9..8104c091ee 100644
--- a/contributor-guide/iceberg-writes.html
+++ b/contributor-guide/iceberg-writes.html
@@ -775,7 +775,7 @@ reserves for metadata columns, so those writes stay on iceberg-java.</p>
the FFI scan of the child’s batches. <code class="docutils literal notranslate"><span class="pre">run_write_task</span></code> builds iceberg-rust’s partitioning writer once
per task, and <code class="docutils literal notranslate"><span class="pre">PartitionWriterBuilder</span></code> builds the rest of the stack once per partition, with that
partition’s Parquet writer properties:</p>
-<div class="highlight-text notranslate"><div class="highlight"><pre><span></span>UnpartitionedWriter | FanoutWriter | ClusteredWriter
+<div class="highlight-text notranslate"><div class="highlight"><pre><span></span>UnpartitionedWriter | FanoutPartitions | ClusteredWriter
-> PartitionWriterBuilder, per partition:
ParquetWriterBuilder -> RollingFileWriterBuilder -> DataFileWriterBuilder
</pre></div>
@@ -822,8 +822,9 @@ held rows then go through <code class="docutils literal notranslate"><span class
<li><p><strong>Field ids and casting.</strong> <code class="docutils literal notranslate"><span class="pre">decorate_batch_with_field_ids</span></code> casts each batch to the
field-id-annotated Arrow schema derived from the Iceberg schema, with <code class="docutils literal notranslate"><span class="pre">safe:</span> <span class="pre">false</span></code>, so a type
mismatch fails the task instead of writing NULLs.</p></li>
-<li><p><strong>Deterministic output order.</strong> iceberg-rust’s <code class="docutils literal notranslate"><span class="pre">FanoutWriter</span></code> closes its writers out of a
-<code class="docutils literal notranslate"><span class="pre">HashMap</span></code>, so the fanout path sorts its <code class="docutils literal notranslate"><span class="pre">DataFile</span></code>s by path before returning them
+<li><p><strong>Deterministic output order.</strong> <code class="docutils literal notranslate"><span class="pre">FanoutPartitions</span></code>, Comet’s version of iceberg-rust’s
+<code class="docutils literal notranslate"><span class="pre">FanoutWriter</span></code>, closes its files in an order that depends on how the rows arrived and on which
+partitions it closed early, so the fanout path sorts its <code class="docutils literal notranslate"><span class="pre">DataFile</span></code>s by path before returning them
(<a class="reference external" href="https://github.com/apache/datafusion-comet/issues/5776">#5776</a>). Manifest order becomes the row
order of an unordered read, so any map iteration that reaches the output needs the same care.</p></li>
</ul>
@@ -833,8 +834,21 @@ hold in memory. It hands <code class="docutils literal notranslate"><span class=
writer counts the bytes that leave memory on their way to storage, and a file reports what it has
written less those. When they leave depends on the storage (<code class="docutils literal notranslate"><span class="pre">StorageWrites</span></code>). After every batch
<code class="docutils literal notranslate"><span class="pre">run_write_task</span></code> resizes the task’s reservation to what the open files report plus the rows each
-<code class="docutils literal notranslate"><span class="pre">PartitionFeed</span></code> holds, for the dictionary choice or for pacing, and a resize the pool refuses fails
-the task. What that figure covers, and what it misses, is described under
+<code class="docutils literal notranslate"><span class="pre">PartitionFeed</span></code> holds, for the dictionary choice or for pacing (<code class="docutils literal notranslate"><span class="pre">InnerWriter::reserve</span></code>). When the
+pool refuses a fanout write, <code class="docutils literal notranslate"><span class="pre">reserve</span></code> writes out and closes the partitions holding the most, in
+their open file and their feed, until the resize succeeds, and <code class="docutils literal notranslate"><span class="pre">run_write_task</span></code> counts them in the
+<code class="docutils literal notranslate"><span class="pre">files_closed_early</span></code> metric. Taking the largest first frees the most memory per file closed, so
+each refusal closes as few files as it can. Each partition’s files report to a child of the task’s
+<code class="docutils literal notranslate"><span class="pre">OpenFileMemory</span></code>, which is how <code class="docutils literal notranslate"><span class="pre">reserve</span></code> finds them. That is why the fanout path uses
+<code class="docutils literal notranslate"><span class="pre">FanoutPartitions</span></code> rather than iceberg-rust’s <code class="docutils literal notranslate"><span class="pre">FanoutWriter</span></code>, which cannot close one partition’s
+writer. A closed partition’s next rows open a new file with the properties its first file used,
+which the partition keeps. A refusal from the pool cannot fail a fanout write: everything the write
+reserves belongs to one of its partitions, so once all of them are closed it reserves nothing, and a
+resize down always succeeds. Only a refusal (<code class="docutils literal notranslate"><span class="pre">ResourcesExhausted</span></code>) makes <code class="docutils literal notranslate"><span class="pre">reserve</span></code> close
+partitions. Any other error from a resize fails the task, such as an exception Spark threw because
+a spill failed or because the task was killed while it waited for memory. An unpartitioned or
+clustered write has no file it can close early, so a resize the pool refuses fails its task. What
+the reserved figure covers, and what it misses, is described under
<a class="reference internal" href="memory_management.html#native-writers"><span class="std std-ref">Native writers</span></a>.</p>
<p><code class="docutils literal notranslate"><span class="pre">FileIO</span></code> comes from <code class="docutils literal notranslate"><span class="pre">load_file_io</span></code> in <code class="docutils literal notranslate"><span class="pre">iceberg_common.rs</span></code>, shared with the native scan. It picks
the storage backend from the data location’s scheme and wires in Comet’s S3 credential bridge when
@@ -996,7 +1010,7 @@ the writer: run the write suites and the Iceberg Spark tests. The pin policy is
</thead>
<tbody>
<tr class="row-even"><td><p><code class="docutils literal notranslate"><span class="pre">CometIcebergWriteActionSuite</span></code></p></td>
-<td><p>End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, a fanout write outgrowing the memory pool, AQE re-planning.</p></td>
+<td><p>End-to-end writes through the split plan and the native writer: parity with iceberg-java, row-level DML, partition evolution, file order, cleanup on task and job failure, writes outgrowing the memory pool, AQE re-planning.</p></td>
</tr>
<tr class="row-odd"><td><p><code class="docutils literal notranslate"><span class="pre">CometIcebergWriteDetectionSuite</span></code></p></td>
<td><p>One case per eligibility rule, accepted and declined.</p></td>
diff --git a/contributor-guide/memory_management.html b/contributor-guide/memory_management.html
index 06d026a120..07e5408ff7 100644
--- a/contributor-guide/memory_management.html
+++ b/contributor-guide/memory_management.html
@@ -957,8 +957,11 @@ less, as overcommit (see <a class="reference internal" href="#the-unified-pools"
<section id="native-writers">
<h3>Native writers<a class="headerlink" href="#native-writers" title="Link to this heading">#</a></h3>
<p>Both native writers reserve what they hold between batches through a single consumer per task,
-<code class="docutils literal notranslate"><span class="pre">ParquetWriterExec[N]</span></code> or <code class="docutils literal notranslate"><span class="pre">IcebergWriteExec[N]</span></code>, resized after every batch. Neither can spill, so
-when the pool refuses a resize the task fails with a <code class="docutils literal notranslate"><span class="pre">CometNativeException</span></code> whose message starts
+<code class="docutils literal notranslate"><span class="pre">ParquetWriterExec[N]</span></code> or <code class="docutils literal notranslate"><span class="pre">IcebergWriteExec[N]</span></code>, resized after every batch. Neither can spill. A
+fanout Iceberg write gives memory back instead, by closing partitions early when the pool refuses a
+resize, as described below, so a refusal never fails it. An exception Spark throws during a resize,
+such as a failed spill or a killed task, still fails the task. When the pool refuses any other
+write’s resize, the task fails with a <code class="docutils literal notranslate"><span class="pre">CometNativeException</span></code> whose message starts
<code class="docutils literal notranslate"><span class="pre">Additional</span> <span class="pre">allocation</span> <span class="pre">failed</span> <span class="pre">for</span></code> and names the consumer. That is a task failure Spark can retry.
Unreserved, the same memory would count only toward the container limit, where exceeding it kills
the executor.</p>
@@ -979,10 +982,13 @@ in memory until the next one is complete or the file closes, so there a file als
it has flushed since its last part was uploaded. At the default row-group size
(<code class="docutils literal notranslate"><span class="pre">write.parquet.row-group-size-bytes</span></code>, 128 MiB) that is the last row group it flushed. The
reservation also covers the rows each partition holds back, first for its dictionary choice and
-then until they fill the 1000-row unit the rolling writer is fed in. It does not cover what
-parquet-rs holds beyond the encoded size: dictionary hash tables, unencoded dictionary indices
-and buffer capacity. Native writes decline Bloom filters today, and neither figure would include
-them.</p></li>
+then until they fill the 1000-row unit the rolling writer is fed in. When the pool refuses a
+fanout write’s resize, the writer writes out and closes the partitions holding the most, file
+and held rows together, until the resize succeeds, so the write ends with more files rather than
+failing. Each file also reports to its partition’s total, which is how the writer finds them.
+The reservation does not cover what parquet-rs holds beyond the encoded size: dictionary hash
+tables, unencoded dictionary indices and buffer capacity. Native writes decline Bloom filters
+today, and neither figure would include them.</p></li>
</ul>
<p>The writers register one consumer per task rather than one per open file because every consumer
registered with <code class="docutils literal notranslate"><span class="pre">fair_unified</span></code> lowers the share of every other consumer in the task. A consumer per
diff --git a/searchindex.js b/searchindex.js
index 12ec626ee5..4659440baa 100644
--- a/searchindex.js
+++ b/searchindex.js
@@ -1 +1 @@
-Search.setIndex({"alltitles": {"!": [[59, "id1"]], "%": [[57, "id1"]], "&": [[47, "id1"]], "*": [[57, "id2"]], "+": [[57, "id3"]], "-": [[57, "id4"]], "/": [[57, "id5"]], "1. Format Your Code": [[44, "format-your-code"]], "1. Install Comet": [[66, "install-comet"], [78, "install-comet"]], "1. Native Operators (nativeExecs map)": [[30, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[44, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[66, "clone-iceberg-and-apply- [...]
\ No newline at end of file
+Search.setIndex({"alltitles": {"!": [[59, "id1"]], "%": [[57, "id1"]], "&": [[47, "id1"]], "*": [[57, "id2"]], "+": [[57, "id3"]], "-": [[57, "id4"]], "/": [[57, "id5"]], "1. Format Your Code": [[44, "format-your-code"]], "1. Install Comet": [[66, "install-comet"], [78, "install-comet"]], "1. Native Operators (nativeExecs map)": [[30, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[44, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[66, "clone-iceberg-and-apply- [...]
\ No newline at end of file
diff --git a/user-guide/latest/iceberg-writes.html b/user-guide/latest/iceberg-writes.html
index 9dff6b4114..2aaec30483 100644
--- a/user-guide/latest/iceberg-writes.html
+++ b/user-guide/latest/iceberg-writes.html
@@ -1162,13 +1162,23 @@ attempt id is embedded in its data file names.</p>
native operators draw on, where iceberg-java’s buffers sit on the JVM heap. A fanout write keeps a
data file open for every partition a task writes to. Each open file holds the row group it is
writing in memory, up to <code class="docutils literal notranslate"><span class="pre">write.parquet.row-group-size-bytes</span></code>, and on S3 or GCS also the last row
-group it flushed, which is uploaded once the next one is complete or the file closes. So a task
-writing to many partitions needs memory in proportion to them. When the pool cannot grant it, the
-task fails with a <code class="docutils literal notranslate"><span class="pre">CometNativeException</span></code> reading <code class="docutils literal notranslate"><span class="pre">Additional</span> <span class="pre">allocation</span> <span class="pre">failed</span> <span class="pre">for</span> <span class="pre">IcebergWriteExec</span></code>
-instead of exceeding the executor’s memory, and Spark retries it like any other task failure. Such a
-write fits in less memory with the fanout writer disabled (<code class="docutils literal notranslate"><span class="pre">write.spark.fanout.enabled=false</span></code>):
-Spark then sorts each task’s rows by partition, and the task keeps one file open at a time. A
-smaller row-group size also helps. Otherwise the write needs a larger <code class="docutils literal notranslate"><span class="pre">spark.memory.offHeap.size</span></code>.</p>
+group it flushed, which is uploaded once the next one is complete or the file closes. Each
+partition also holds the rows that have not reached its file yet. So a task writing to many
+partitions needs memory in proportion to them. When the pool cannot grant it, the task writes out
+and closes the partitions holding the most memory until what is left fits, and a closed
+partition’s next rows open a new file. The write then finishes with more, smaller files than
+iceberg-java’s would. When every partition keeps getting rows, a task given an eighth of the memory
+its partitions need writes about four files per partition instead of one, however many partitions
+it writes to. The <code class="docutils literal notranslate"><span class="pre">files</span> <span class="pre">closed</span> <span class="pre">early</span> <span class="pre">to</span> <span class="pre">free</span> <span class="pre">memory</span></code> metric of its <code class="docutils literal notranslate"><span class="pre">CometIcebergWrite</span></code> operator
+counts the files it closed early. Disabling the fanout writer (<code class="docutils literal notranslate"><span class="pre">write.spark.fanout.enabled=false</span></code>)
+avoids the extra files: Spark then sorts each task’s rows by partition, and the task keeps one file
+open at a time. A larger <code class="docutils literal notranslate"><span class="pre">spark.memory.offHeap.size</span></code> also helps, and Iceberg’s <code class="docutils literal notranslate"><span class="pre">rewrite_data_files</span></code>
+procedure compacts small files once they are written.</p>
+<p>A write that keeps one file open, unpartitioned or clustered, has no partition to close. When that
+file outgrows the pool, the task fails with a <code class="docutils literal notranslate"><span class="pre">CometNativeException</span></code> reading
+<code class="docutils literal notranslate"><span class="pre">Additional</span> <span class="pre">allocation</span> <span class="pre">failed</span> <span class="pre">for</span> <span class="pre">IcebergWriteExec</span></code> instead of exceeding the executor’s memory, and
+Spark retries it like any other task failure. A smaller <code class="docutils literal notranslate"><span class="pre">write.parquet.row-group-size-bytes</span></code> or a
+larger <code class="docutils literal notranslate"><span class="pre">spark.memory.offHeap.size</span></code> lets such a write fit.</p>
<p>Partial results are never committed. The commit set is exactly the commit messages returned by
successful tasks — a failed task contributes none — and if the job fails, the driver-side
commit operator aborts without committing anything. A failed task attempt also deletes the
@@ -1260,6 +1270,11 @@ independent size estimates, so nothing bounds how far apart the two writers’ r
they may cross the target several grid steps apart, and the resulting files can differ in row
count by an arbitrary number of 1000-row blocks. Do not rely on file-layout parity between the
two writers; rely only on each file rolling on its own 1000-row boundary.</p></li>
+<li><p>A fanout write that the memory pool cannot hold closes partitions before the task ends (see
+<a class="reference internal" href="#failure-handling">Failure handling</a>), where iceberg-java’s fanout writer keeps every file open
+until then. So a partition can have more, smaller files than iceberg-java writes, and a file
+closed early ends off the 1000-row grid, as the last file of a task does. A partition closed
+before its rows filled a first page makes its dictionary choice from the rows it has.</p></li>
<li><p>A fanout write lists a task’s data files in file-path order, where iceberg-java lists them in
its own <code class="docutils literal notranslate"><span class="pre">StructLikeMap</span></code> iteration order. Both are stable across runs, and neither is a
documented ordering, but the manifest entry order becomes the scan-task order and so the row
@@ -1267,10 +1282,9 @@ order of an unordered <code class="docutils literal notranslate"><span class="pr
commit gives each file’s rows, so the same rows can get different <code class="docutils literal notranslate"><span class="pre">_row_id</span></code> values from the two
writers. The ids are unique either way, and across tasks iceberg-java’s own assignment already
depends on the order in which the tasks finish. Only the sorted order is reproducible on the
-native path: iceberg-rust’s <code class="docutils literal notranslate"><span class="pre">FanoutWriter</span></code> closes its per-partition writers out of a <code class="docutils literal notranslate"><span class="pre">HashMap</span></code>,
-which under Rust’s per-process <code class="docutils literal notranslate"><span class="pre">RandomState</span></code> would otherwise give a different order on every
-run. Clustered and unpartitioned writes append in creation order on both paths and are
-unaffected.</p></li>
+native path: the order the native writer closes its files in depends on the order rows arrive in
+and on which partitions it closes early to free memory. Clustered and unpartitioned writes append
+in creation order on both paths and are unaffected.</p></li>
<li><p>Compressed page bytes are implementation-defined: the codec and any explicit level are
translated, but parquet-rs and parquet-mr embed different encoder implementations and
defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org