fix: list a fanout Iceberg write's data files in a stable order

[andygrove]

Which issue does this PR close?

Closes #5776.

Rationale for this change

iceberg-rust's FanoutWriter keeps its per-partition writers in a HashMap and closes them by iterating it directly:

async fn close(mut self) -> Result<O> {
    for (_, mut writer) in self.partition_writers {
        self.output.extend(writer.close().await?);
    }
    Ok(O::from_iter(self.output))
}

Under Rust's per-process RandomState that gives a different DataFile order on every run. That order becomes the manifest entry order, which becomes the scan-task order, which becomes the row order of an unordered SELECT *. Iceberg's own TestMetadataTablesWithPartitionEvolution.testPartitionColumnNamedPartition compares such a SELECT * positionally against what iceberg-java wrote, so it failed whenever the two disagreed.

Measured on Spark 4.1 + Iceberg 1.11 by writing eight partitions through the fanout writer and printing the manifest order on three consecutive runs:

run 1: partition=1, partition=2, partition=5, partition=0, partition=7, partition=4, partition=3, partition=6
run 2: partition=6, partition=7, partition=4, partition=5, partition=0, partition=3, partition=2, partition=1
run 3: partition=2, partition=0, partition=4, partition=5, partition=7, partition=3, partition=6, partition=1

The clustered writer produced partition=0 .. partition=7 on every run.

Two notes on the issue as filed. The failing parameter set includes ORC and AVRO, which looked like evidence the Parquet writer could not be involved; it is not, because that test does not use the class's createTable helper and so writes Parquet whatever the fileFormat parameter says. And the test only has two partitions, so an unfixed writer already passes about half the time, which is why only a subset of the matrix reported failures.

What changes are included in this PR?

• Sort the fanout writer's output by file path in InnerWriter::close.

  It cannot be creation order instead: RecordBatchPartitionSplitter splits a batch through a map as well, so which partition is written first, and therefore which file-name counter it gets, is itself unstable. Path order sorts by partition directory and then by that counter within a partition, and needs nothing from the write order. The clustered and unpartitioned writers append in creation order and are already deterministic, so only the fanout arm sorts.

• Record the behaviour under accepted divergences in iceberg-writes.md. This is determinism, not parity: iceberg-java's fanout writer iterates its own StructLikeMap, so the two agree only where that order and path order coincide, which they do for the ascending partition values the upstream test uses.

Part of #5649, and one of the two remaining failure buckets blocking #5644.

How are these changes tested?

A Rust test (fanout_write_returns_data_files_in_a_deterministic_order) and a Scala test (a fanout write lists its data files in a stable order), both over eight partitions rather than the upstream test's two, so an unfixed writer lands in the right order by luck 1 time in 8! rather than half the time. Both were confirmed to fail three times out of three on the unfixed writer and pass three times out of three with the fix.

The end-to-end scenario from the issue was also swept over spark.sql.shuffle.partitions in {1, 2, 4, 200} against both writers: parts=2 with the native writer was the combination that reproduced the reordering, and all eight combinations now return the expected rows on three consecutive runs.

CometIcebergWriteActionSuite (62), CometIcebergWriteDetectionSuite (48), CometIcebergSystemFunctionSuite (13), CometIcebergRewriteActionSuite (5) and the native iceberg tests (53) pass.

🤖 Generated with Claude Code

https://claude.ai/code/session_018sqhRddDS2gFfJRT21QDsE

[sunchao]

Correctness

Summary and implementation

Reviewed b5069564b99193cbafb1ddb3ed28a040183b28e0 against 424c31aa79d13fddf743ffa29bae3c6f146e6c5e. No verified P1/P2 findings. Approving the implementation. The three-file change addresses unstable file-list ordering from iceberg-rust's fanout writer. It sorts the completed task's DataFile vector by file path, adds Rust and Spark tests over eight partitions, and documents the ordering difference from iceberg-java. Clustered and unpartitioned writers keep their existing paths. The existing split-writer and native-write flags remain off by default.

The pinned iceberg-rust source confirms both sources of instability: the fanout writer closes a HashMap of partition writers, and the batch splitter also groups through a HashMap. File counters are therefore unsuitable for deciding the order of different partition directories. Sorting after successful close avoids changing partition routing or file creation. The returned files, their contents and their metrics are preserved, and close errors still propagate before the sort. Empty and single-file output require no special case.

Spark compatibility and behavior

The sorted vector is written into the task's transport manifest in order. JVM decoding, metric rebuilding and TaskCommit construction preserve that order. In Iceberg 1.11, SparkWrite.files(messages) uses DataFileSet, whose underlying WrapperSet is insertion-ordered, so commit collection does not discard the new ordering. Maintained Spark 3.5/4.0 collects partition results without imposing a SQL sort. The canonical SQL reference reserves a total ordering guarantee for ORDER BY.

The supported claim is therefore file-path order within a fanout task, not a general promise of row ordering or parity with Java's partition-map iteration. Lexical path order is also not a typed partition-key sort. The new fixture observes ascending rows for its eight one-file partitions under the test configuration. It does not establish that order across arbitrary task distributions, file rolling, manifest rewrites or later scan planning. The accepted-divergence documentation should be read within that task-local scope. No expression, null, ANSI, overflow, partition transform, write-admission or commit-transaction semantics change here.

Validation and limits

The Rust CI job passes the new real-writer integration test, with 1,259 passed and five skipped overall. The new Scala test passes on Spark 3.4, 3.5, 4.0 and 4.1. Its helper requires a native writer plan, exactly one new snapshot and the expected rows. Spark 4.2 cancels this test because Iceberg is absent from its classpath. The four passing jobs cover native/JNI execution rather than only a test double.

All inspected jobs checked out merge ccfd9d8b, whose parents are this base and head and whose entire source tree equals head. The native producer and scans consumers record the same artifact ID and SHA256 archive digest, also matched against artifact metadata. No binary payload or local product build was needed. The upstream Iceberg 1.11 job passes six instances of the issue's test, but its unchanged harness leaves the native-write flag off by default, so those passes are not additional native-write qualification. The author's repeated before/after runs and shuffle-partition sweep remain author reports. Maintained Spark 3.4/4.1 source gaps remain despite applicable CI execution.

At 2026-09-09T19:47:53.218226+00:00, checks show 59 successful, nine skipped, one queued and one failed. The Delta build gate fails because the contrib-enabled library is 1,519,296,760 bytes, smaller than the default library's 1,519,362,160 bytes. Its default dependency, class, service and symbol-exclusion checks passed. The later contrib-symbol assertion did not run. This identifies the failed gate without claiming that CI is green or that the failure was reproduced on base.

Performance

The additional work occurs once when a fanout task closes. sort_unstable_by sorts the existing vector in place and compares borrowed path strings. It adds no scratch allocation, file I/O, row rewriting or cloned path keys. The sort is O(F log F) comparisons for F output files, with comparison cost depending on shared path prefixes. Clustered and unpartitioned writes incur no new work. There is no measured throughput claim or new expression kernel here, and the eight-partition correctness test is not a benchmark. I found no evidence of a material performance regression that warrants a P2.

Design

Normalizing the list at InnerWriter::close is a small, well-placed change: it uses the complete file set and leaves the upstream writer lifecycle intact. Replacing the dependency's map or sorting every incoming batch would affect more code and the write path itself. The chosen boundary also keeps commit message order and abort behavior unchanged. The tests complement each other: Rust pins the returned list, while Scala checks the committed metadata and observed readback with native writing enabled. They establish reproducibility for those fixtures without making Java ordering a compatibility requirement.

Abstraction & complexity

The change adds no new type, configuration, shared state or indirection. A short branch around the existing close result is enough. The larger comments explain why creation order is insufficient and why only fanout needs normalization. Existing helpers continue to own path rendering, manifest transport and commit construction. No additional abstraction or simplification is required before merge.

[andygrove]

Merged. Thanks @sunchao!