From b5069564b99193cbafb1ddb3ed28a040183b28e0 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Wed, 9 Sep 2026 11:18:08 -0600 Subject: [PATCH] fix: list a fanout Iceberg write's data files in a stable order iceberg-rust's `FanoutWriter` keeps its per-partition writers in a `HashMap` and closes them by iterating it, so the `DataFile` order a task returned followed Rust's per-process `RandomState` and differed on every run. That order is the manifest entry order, which is the scan-task order, which is the row order of an unordered `SELECT *`. Iceberg's `TestMetadataTablesWithPartitionEvolution.testPartitionColumnNamedPartition` compares such a `SELECT *` positionally against what iceberg-java wrote, and failed whenever the two disagreed. Sort the fanout writer's output by file path. 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. 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. Recorded under accepted divergences. Closes #5776 --- .../user-guide/latest/iceberg-writes.md | 7 ++ .../src/execution/operators/iceberg_write.rs | 74 ++++++++++++++++++- .../comet/CometIcebergWriteActionSuite.scala | 44 +++++++++++ 3 files changed, 124 insertions(+), 1 deletion(-) diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index 62a4df1f128..f09cd80be70 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -260,6 +260,13 @@ a data file but not what any reader computes from it: unescaped; Comet uses the 1.8+ spelling on every profile. Distinct partition values still get distinct directories in all cases, and no reader parses these names — files are resolved through committed manifests. Iceberg deprecated float and double partitioning in 1.3. +- 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 diff --git a/native/core/src/execution/operators/iceberg_write.rs b/native/core/src/execution/operators/iceberg_write.rs index 297d9a0c747..3dd4c27768e 100644 --- a/native/core/src/execution/operators/iceberg_write.rs +++ b/native/core/src/execution/operators/iceberg_write.rs @@ -441,7 +441,27 @@ impl InnerWriter { use iceberg::writer::partitioning::PartitioningWriter; match self { InnerWriter::Unpartitioned(w) => w.close().await.map_err(iceberg_err), - InnerWriter::Fanout(w) => w.close().await.map_err(iceberg_err), + InnerWriter::Fanout(w) => { + 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(w) => w.close().await.map_err(iceberg_err), } } @@ -1135,6 +1155,58 @@ 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 = PartitionSpec::builder(Arc::new(schema.clone())) + .with_spec_id(1) + .add_partition_field("region", "region", Transform::Identity) + .unwrap() + .build() + .unwrap(); + 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], ®ions)], + ) + .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(); diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 662d7cb3593..be26ab0f700 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/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) }