diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index fd89e44e141..20b9635f59b 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/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 diff --git a/native/core/src/execution/operators/iceberg_write.rs b/native/core/src/execution/operators/iceberg_write.rs index ac1e837714a..3a988e98389 100644 --- a/native/core/src/execution/operators/iceberg_write.rs +++ b/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], ®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 f1738cd8b6b..0c6affcd30a 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) }