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

docs/source/user-guide/latest/iceberg-writes.md

@@ -271,6 +271,13 @@ 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 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
+  order of an unordered `SELECT *`. 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.
 - 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

native/core/src/execution/operators/iceberg_write.rs

@@ -516,7 +516,25 @@ impl InnerWriter {
                         w.write(key, rest).await.map_err(iceberg_err)?;
                     }
                 }
-                w.close().await.map_err(iceberg_err)
+                let mut data_files = w.close().await.map_err(iceberg_err)?;
+                // `FanoutWriter` holds its per-partition writers in a `HashMap` and `close`
+                // iterates it directly, so the order it returns follows Rust's per-process
+                // `RandomState` and differs on every run. That order becomes the manifest entry
+                // order, then the scan-task order, then the row order of an unordered
+                // `SELECT *` -- which Iceberg's own
+                // `TestMetadataTablesWithPartitionEvolution.testPartitionColumnNamedPartition`
+                // compares positionally against what iceberg-java wrote
+                // (apache/datafusion-comet#5776).
+                //
+                // Sorting by path is what makes the task's output reproducible. 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, 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 this arm sorts.
+                data_files.sort_unstable_by(|a, b| a.file_path().cmp(b.file_path()));
+                Ok(data_files)
             }
             InnerWriter::Clustered(mut w, live) => {
                 if let Some((key, mut pacer)) = live {
@@ -1578,6 +1596,53 @@ mod tests {
             assert_eq!(total, 4);
         }
 
+        /// A fanout task's data files must come back in a deterministic order.
+        ///
+        /// iceberg-rust's `FanoutWriter` keeps its per-partition writers in a `HashMap` and
+        /// `close` iterates it directly, so the `DataFile` order it returns follows Rust's
+        /// per-process `RandomState` -- a different 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, and fails when the
+        /// two disagree. See https://github.com/apache/datafusion-comet/issues/5776.
+        ///
+        /// Eight partitions rather than two: a random permutation matching sorted order by luck
+        /// is 1 in 8!.
+        #[tokio::test]
+        async fn fanout_write_returns_data_files_in_a_deterministic_order() {
+            let temp_dir = TempDir::new().unwrap();
+            let data_location = format!("file://{}", temp_dir.path().display());
+            let schema = iceberg_user_schema();
+            let spec = identity_region_spec(&schema);
+            let common = common(
+                data_location,
+                serde_json::to_string(&spec).unwrap(),
+                serde_json::to_string(&schema).unwrap(),
+                ProtoIcebergWriterMode::IcebergWriterFanout,
+            );
+
+            let regions = ["r0", "r1", "r2", "r3", "r4", "r5", "r6", "r7"];
+            let data_files = run(
+                common,
+                schema,
+                spec,
+                ProtoIcebergWriterMode::IcebergWriterFanout,
+                vec![batch(&[0, 1, 2, 3, 4, 5, 6, 7], &regions)],
+            )
+            .await
+            .unwrap();
+
+            assert_eq!(data_files.len(), regions.len());
+            let paths: Vec<&str> = data_files.iter().map(|f| f.file_path()).collect();
+            let mut sorted = paths.clone();
+            sorted.sort_unstable();
+            assert_eq!(
+                paths, sorted,
+                "fanout data files are not in file-path order"
+            );
+        }
+
         #[tokio::test]
         async fn clustered_partitioned_write_handles_sorted_input() {
             let temp_dir = TempDir::new().unwrap();

spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala

@@ -2165,6 +2165,50 @@ class CometIcebergWriteActionSuite
    * (same as the JVM-path assertion -- AQE re-planning never duplicates commits) AND at least one
    * [[CometIcebergWriteExec]] appears in some captured plan AND the resulting row set matches.
    */
+  // https://github.com/apache/datafusion-comet/issues/5776. iceberg-rust's `FanoutWriter` keeps
+  // its per-partition writers in a `HashMap` and closes them by iterating it, so the data-file
+  // order a task returned followed Rust's per-process `RandomState`. That order is the manifest
+  // entry order, which is the scan-task order, which is the row order of an unordered
+  // `SELECT *` -- and Iceberg's own
+  // `TestMetadataTablesWithPartitionEvolution.testPartitionColumnNamedPartition` compares such a
+  // `SELECT *` positionally against what iceberg-java wrote.
+  //
+  // What is asserted here is determinism, not parity: iceberg-java's fanout writer iterates its
+  // own `StructLikeMap`, so the two writers only agree where that map order and path order happen
+  // to coincide -- which they do for the ascending partition values the upstream test uses. See
+  // the fanout entry under accepted divergences in `iceberg-writes.md`.
+  //
+  // Eight partitions rather than the two that test uses: an unfixed writer lands in path order by
+  // luck 1 time in 8!, where with two it would pass half the time.
+  test("native acceleration: a fanout write lists its data files in a stable order") {
+    assumeNativeAcceleration()
+    withIcebergCatalog { warehouseDir =>
+      val fanout = Some("'write.spark.fanout.enabled'='true'")
+      createTable(
+        warehouseDir,
+        "fanout_order",
+        partitionSpec = "PARTITIONED BY (region)",
+        properties = fanout)
+      val values = (0 until 8).map(i => s"($i, 'r$i', $i.5)").mkString(", ")
+
+      assertNativeWriteEngages("fanout_order", 0 until 8) {
+        spark.sql(s"INSERT INTO $catalog.$ns.fanout_order VALUES $values")
+      }
+
+      val paths = spark
+        .sql(s"SELECT file_path FROM $catalog.$ns.fanout_order.files")
+        .collect()
+        .toSeq
+        .map(_.getString(0))
+      assert(paths.size == 8, s"expected one file per partition, got $paths")
+      assert(paths == paths.sorted, s"fanout data files are not listed in path order: $paths")
+
+      // The manifest order is what an unordered read comes back in, so it is stable too.
+      val ids = spark.sql(s"SELECT id FROM $catalog.$ns.fanout_order").collect().toSeq
+      assert(ids.map(_.getInt(0)) == (0 until 8), s"unordered read: ${ids.mkString(", ")}")
+    }
+  }
+
   private def assertNativeWriteEngages(tableName: String, expectedIds: Seq[Int])(
       action: => Unit): Unit = {
     val snapshot = withNativeEnabled { captureWrite(tableName)(action) }