From 537488876ef45fe718900e3da9d4a88376456f77 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 10 Oct 2026 07:14:44 -0600 Subject: [PATCH 1/5] feat: bump iceberg-rust and write float and double identity partitions natively apache/iceberg-rust#3327 gives `PrimitiveLiteral` an equality and ordering that follow Java's `Float.compare` / `Double.compare`, so iceberg-rust's writers keep -0.0 and 0.0 partitions apart, as iceberg-java does. Move the pin from af1da4c to 1f3bc34, the merge commit of that PR, and drop the fallback that #6531 added for float and double partition fields (#6138). Two other upstream changes in the range need code changes: - apache/iceberg-rust#3145 makes `FileScanTaskDeleteFile`'s fields private behind a validating builder, and holds the deletion-vector offset and size as u64. The planner builds delete files through the builder and rejects negative coordinates, and the scan sizes a delete file by rebuilding it. - apache/iceberg-rust#3323 fixes iceberg-rust's `day` for pre-epoch timestamps (apache/iceberg-rust#3315). Comet computes those partition values itself, so only the canary that pins iceberg-rust's values changes. --- .../review-comet-iceberg-write-pr/SKILL.md | 4 +- .../contributor-guide/iceberg-writes.md | 5 +- .../user-guide/latest/iceberg-writes.md | 13 +- native/Cargo.lock | 114 ++++++++++++++--- native/Cargo.toml | 4 +- .../operators/iceberg_partition_value.rs | 13 +- .../src/execution/operators/iceberg_scan.rs | 100 +++++++++++---- native/core/src/execution/planner.rs | 115 +++++++++++++++--- .../comet/iceberg/IcebergReflection.scala | 40 ------ .../operator/CometIcebergNativeWrite.scala | 24 ---- .../comet/CometIcebergWriteActionSuite.scala | 90 ++++++++------ .../CometIcebergWriteDetectionSuite.scala | 103 +++------------- 12 files changed, 360 insertions(+), 265 deletions(-) diff --git a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md index a835f45441a..8b807696ffb 100644 --- a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md +++ b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md @@ -107,8 +107,8 @@ only the latest. that starts trusting iceberg-rust's metrics, or adds a native-tracked value beyond float and double NaN counts and bounds, needs a strong reason. - [ ] **Value semantics match, not only formats.** iceberg-rust compares and hashes partition values - with its own rules: `OrderedFloat` treats `-0.0` and `0.0` as equal where Java's - `Float.compare` does not + with its own rules: until apache/iceberg-rust#3327 its `OrderedFloat` equality treated + `-0.0` and `0.0` as equal where Java's `Float.compare` does not ([#6138](https://github.com/apache/datafusion-comet/issues/6138)). Check equality, hashing, ordering and rendering for float, double, timestamp, timestamptz, binary and decimal. - [ ] **Timestamp partition values are UTC.** Iceberg's `years`, `months`, `days` and `hours` diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index 1ef7539329a..c0aaff3550a 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -395,8 +395,9 @@ Useful places to look when checking parity: - `CometIcebergWriteActionSuite` writes the same data through both writers into sibling tables and compares rows, `readable_metrics` and partition paths. - iceberg-rust's own code, at the pinned revision. iceberg-rust makes different choices from - iceberg-java in places that matter to the table's contents, for example grouping partition keys - with `OrderedFloat`, which treats `-0.0` and `0.0` as equal + iceberg-java in places that matter to the table's contents. For example, until + [apache/iceberg-rust#3327](https://github.com/apache/iceberg-rust/pull/3327) it grouped + partition keys with `OrderedFloat`, which treats `-0.0` and `0.0` as equal ([#6138](https://github.com/apache/datafusion-comet/issues/6138)). Check how iceberg-rust compares, hashes and renders values, not only what it writes. diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index a04f6be61cf..5b2ca522026 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -203,7 +203,7 @@ A write is eligible only when ALL of the following hold: | resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` | | Hadoop S3A settings for an `s3` / `s3a` data location | only `fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`, `endpoint.region`, and `path.style.access`, including their `fs.s3a.bucket..*` forms; any other effective `fs.s3a.*` setting falls back | | Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location | the S3 endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`, `kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain settings parsed by the pinned iceberg-rust version, plus Comet's credential-provider class and built-in web-identity properties. When a custom provider is configured, its vendor-owned `s3.*` / `client.*` properties are also forwarded; unsupported Iceberg-defined S3 settings still fall back | -| partition spec | any, except an identity partition on a `float` or `double` column, or a `void` field whose source column was dropped beside a live field (see below) | +| partition spec | any, except a `void` field whose source column was dropped beside a live field (see below) | | column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) and the v3 types `variant`, `unknown`, `timestamp_ns`, `geometry` and `geography` | Within the namespaces that shape data-file bytes — `write.parquet.*` and `parquet.*` — @@ -254,13 +254,6 @@ native write. If Iceberg's AWS property classes cannot be loaded, vendor `s3.*` fall back too and planning still completes. The fall-back reason reports only sorted property names, never their values, so credentials and tokens do not enter EXPLAIN or plan logs. -An identity partition on a `float` or `double` column falls back. iceberg-rust compares float -partition values with an equality that treats `-0.0` and `0.0` as one value, so the native writer -would put rows with either value in the same partition, where iceberg-java writes two. A read -that prunes on the other value's partition would then miss rows. The fall-back stays until -iceberg-rust distinguishes the two values -([apache/iceberg-rust#3325](https://github.com/apache/iceberg-rust/issues/3325)). - On a format-version 3 table with Iceberg 1.10 or newer, a write that rewrites existing rows falls back. Copy-on-write `DELETE`, `UPDATE` and `MERGE` and `rewrite_data_files` write the row lineage columns `_row_id` and `_last_updated_sequence_number` into the new data files, so that rewritten @@ -391,9 +384,7 @@ a data file but not what any reader computes from it: (iceberg-java names files `---`; iceberg-rust uses a process-local counter). - Partition directory names match iceberg-java 1.8+'s `PartitionSpec.partitionToPath` for every - partition type the native writer accepts. Identity partitions on `float` and `double` columns - fall back (see [Native Parquet write eligibility](#native-parquet-write-eligibility)), so - iceberg-java names those directories itself. On Iceberg 1.5.x, + partition type the native writer accepts. On Iceberg 1.5.x, which the Spark 3.4 profile pins, iceberg-java itself spelled `timestamp` and `timestamptz` directories with `LocalDateTime.toString()` / `OffsetDateTime.toString()` (`ts=1969-12-31T23:59:58.500Z`) and left the partition field name unescaped; Comet uses the diff --git a/native/Cargo.lock b/native/Cargo.lock index c73688fe1d8..a49297204a1 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -75,6 +75,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "aliasable" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "250f629c0161ad8107cf89319e990051fae62832fd343083bea452d93e2205fd" + [[package]] name = "aligned-vec" version = "0.6.4" @@ -178,26 +184,26 @@ checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" [[package]] name = "apache-avro" -version = "0.21.0" +version = "0.22.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf" +checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c" dependencies = [ "bigdecimal", "bon", "crc32fast", - "digest 0.10.7", + "digest 0.11.3", "log", - "miniz_oxide 0.8.9", + "miniz_oxide 0.9.1", "num-bigint 0.4.8", + "ouroboros", "quad-rand", - "rand 0.9.5", + "rand 0.10.2", "regex-lite", "serde", "serde_bytes", "serde_json", "snap", - "strum", - "strum_macros", + "strum 0.28.0", "thiserror 2.0.20", "uuid", "zstd 0.13.3", @@ -1429,7 +1435,7 @@ version = "4.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d012d2b9d65aca7f18f4d9878a045bc17899bba951561ba5ec3c2ba1eed9a061" dependencies = [ - "heck", + "heck 0.5.0", "proc-macro2", "quote", "syn 3.0.5", @@ -3382,6 +3388,12 @@ dependencies = [ "log", ] +[[package]] +name = "heck" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8" + [[package]] name = "heck" version = "0.5.0" @@ -3581,7 +3593,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3" +source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" dependencies = [ "aes-gcm", "anyhow", @@ -3625,7 +3637,7 @@ dependencies = [ "serde_json", "serde_repr", "serde_with", - "strum", + "strum 0.27.2", "tokio", "tracing", "typed-builder", @@ -3639,7 +3651,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3" +source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" dependencies = [ "proc-macro2", "quote", @@ -3649,7 +3661,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=af1da4c5bc86178c38c1c6db0578bdd9fb7021e3#af1da4c5bc86178c38c1c6db0578bdd9fb7021e3" +source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" dependencies = [ "anyhow", "async-trait", @@ -4874,6 +4886,30 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "ouroboros" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e0f050db9c44b97a94723127e6be766ac5c340c48f2c4bb3ffa11713744be59" +dependencies = [ + "aliasable", + "ouroboros_macro", + "static_assertions", +] + +[[package]] +name = "ouroboros_macro" +version = "0.18.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c7028bdd3d43083f6d8d4d5187680d0d3560d54df4cc9d752005268b41e64d0" +dependencies = [ + "heck 0.4.1", + "proc-macro2", + "proc-macro2-diagnostics", + "quote", + "syn 2.0.119", +] + [[package]] name = "outref" version = "0.5.2" @@ -5296,6 +5332,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "proc-macro2-diagnostics" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af066a9c399a26e020ada66a034357a868728e72cd426f3adcd35f80d88d88c8" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", + "version_check", + "yansi", +] + [[package]] name = "procfs" version = "0.18.0" @@ -5336,7 +5385,7 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ - "heck", + "heck 0.5.0", "itertools 0.14.0", "log", "multimap", @@ -6470,6 +6519,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + [[package]] name = "str_stack" version = "0.1.1" @@ -6488,7 +6543,16 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf" dependencies = [ - "strum_macros", + "strum_macros 0.27.2", +] + +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros 0.28.0", ] [[package]] @@ -6497,7 +6561,19 @@ version = "0.27.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7" dependencies = [ - "heck", + "heck 0.5.0", + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck 0.5.0", "proc-macro2", "quote", "syn 2.0.119", @@ -6587,7 +6663,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", "windows-sys 0.61.2", @@ -7492,6 +7568,12 @@ version = "0.13.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" +[[package]] +name = "yansi" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cfe53a6657fd280eaa890a3bc59152892ffa3e30101319d168b781ed6529b049" + [[package]] name = "yoke" version = "0.8.3" diff --git a/native/Cargo.toml b/native/Cargo.toml index dac17e5a289..17b53a6771a 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -82,8 +82,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false } # Pinned to a revision rather than a release because no published iceberg-rust # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain # version requirement once a release ships on arrow 59. Bump policy: #5645. -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc86178c38c1c6db0578bdd9fb7021e3" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "af1da4c5bc86178c38c1c6db0578bdd9fb7021e3", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" [profile.release] diff --git a/native/core/src/execution/operators/iceberg_partition_value.rs b/native/core/src/execution/operators/iceberg_partition_value.rs index 970297283a7..f6838a39a8d 100644 --- a/native/core/src/execution/operators/iceberg_partition_value.rs +++ b/native/core/src/execution/operators/iceberg_partition_value.rs @@ -39,7 +39,7 @@ use iceberg::{Error, ErrorKind, Result}; /// `timestamp`, or `timestamptz` source, which go through Comet's `iceberg_years` / /// `iceberg_months` / `iceberg_days` / `iceberg_hours` kernels instead: the ones the sort in front /// of a clustered write runs, pinned against iceberg-java's `DateTimeUtil` over the whole domain. -/// iceberg-rust's transforms differ from iceberg-java's in three ways: +/// iceberg-rust's transforms differ from iceberg-java's in two ways: /// /// - `year` and `month` split the calendar with Arrow's `date_part`, which returns NULL for /// anything `chrono` cannot represent -- past year 262142 -- whereas iceberg-java goes through @@ -49,9 +49,6 @@ use iceberg::{Error, ErrorKind, Result}; /// - All four floor a pre-epoch timestamp that lies exactly 999999 microseconds into a unit, which /// iceberg-java puts in the unit before, so `1969-01-01T00:00:00.999999` belongs in the 1968 /// partitions (apache/datafusion-comet#6426). -/// - `day` moves a timestamp from the last second of a day before 1969-12-31 into the next day, -/// unless its microsecond of second is 0 or 999999: it takes the whole seconds with a truncating -/// division and the microseconds with a flooring one (apache/iceberg-rust#3315). /// /// A partition value that differs from the sort key can also fail a clustered write, which rejects /// a row whose partition it has already closed. Everywhere else the two implementations agree, so @@ -393,7 +390,8 @@ mod tests { Some(-31_535_999_000_001), // 1969-12-31T23:00:00.999999, where that moves only the hour. Some(-3_599_000_001), - // 1969-12-30T23:59:59.5 and 1969-12-30T23:59:59.999998. + // 1969-12-30T23:59:59.5 and 1969-12-30T23:59:59.999998, which iceberg-rust's `day` + // moved into 1969-12-31 until apache/iceberg-rust#3315 was fixed. Some(-86_400_500_000), Some(-86_400_000_002), None, @@ -419,7 +417,7 @@ mod tests { ( Transform::Day, day([-366, -1, -2, -2]), - day([-365, -1, -1, -1]), + day([-365, -1, -2, -2]), ), ( Transform::Hour, @@ -438,8 +436,7 @@ mod tests { for (i, (transform, java, rust)) in cases.iter().enumerate() { let label = format!("{transform} of {}", source.data_type()); assert_eq!(&comet[i], java, "{label}"); - // The reason Comet computes these: iceberg-rust floors the first two rows, and its - // `day` moves the last two into 1969-12-31 (apache/iceberg-rust#3315). If this + // The reason Comet computes these: iceberg-rust floors the first two rows. If this // starts failing, iceberg-rust's transforms have changed and delegating needs // another look. assert_eq!(&iceberg_rust[i], rust, "iceberg-rust's {label}"); diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index 9f465329ca0..ac30194b3db 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -272,14 +272,14 @@ impl IcebergScanExec { // Delete-file sizes are never serialized, so they always arrive as 0. If we ever // trust manifest sizes (pending the unreleased apache/iceberg#12554 fix), skip // already-sized files here instead of asserting. - debug_assert_eq!(delete.file_size_in_bytes, 0); + debug_assert_eq!(delete.file_size_in_bytes(), 0); // A deletion vector is range-read from content_offset, and iceberg-rust consults // file_size_in_bytes only on the Parquet delete path. Statting the Puffin file // would be one HEAD per file per Spark partition for a value nothing reads. - if delete.file_format == DataFileFormat::Puffin { + if delete.file_format() == DataFileFormat::Puffin { continue; } - needed.insert(delete.file_path.clone()); + needed.insert(delete.file_path().to_string()); } } if needed.is_empty() { @@ -336,19 +336,41 @@ impl IcebergScanExec { let deletes = task .deletes() .iter() - .map(|delete| { - let mut delete = delete.clone(); - if let Some(&size) = size_map.get(&delete.file_path) { - delete.file_size_in_bytes = size; - } - delete + .map(|delete| match size_map.get(delete.file_path()) { + Some(&size) => Self::sized_delete_file(delete, size), + None => Ok(delete.clone()), }) - .collect::>(); + .collect::, Error>>()?; *task = Self::rebuild_task_with_deletes(task, deletes)?; } Ok(()) } + /// Copies a delete file with its size set to `size`. + /// + /// iceberg-rust 1f3bc34 made `FileScanTaskDeleteFile`'s fields private too, so, as for the + /// task in [`Self::rebuild_task_with_deletes`], the copy goes through the builder, forwarding + /// every other field through its public accessors. The builder repeats the delete file's + /// validation on `build()`. + fn sized_delete_file( + delete: &FileScanTaskDeleteFile, + size: u64, + ) -> Result { + FileScanTaskDeleteFile::builder() + .with_file_path(delete.file_path().to_string()) + .with_file_size_in_bytes(size) + .with_file_type(delete.file_type()) + .with_file_format(delete.file_format()) + .with_partition_spec_id(delete.partition_spec_id()) + .with_equality_ids(delete.equality_ids().map(<[i32]>::to_vec)) + .with_referenced_data_file(delete.referenced_data_file().map(str::to_string)) + .with_content_offset(delete.content_offset()) + .with_content_size_in_bytes(delete.content_size_in_bytes()) + .with_record_count(delete.record_count()) + .with_key_metadata(delete.key_metadata().map(Box::from)) + .build() + } + /// Rebuilds a [`FileScanTask`] carrying a new set of delete files. /// /// `FileScanTask::deletes` is private with no mutator, so the only way to change a task's @@ -690,19 +712,42 @@ mod tests { } fn delete_file(path: &str) -> FileScanTaskDeleteFile { - FileScanTaskDeleteFile { - file_path: path.to_string(), - file_type: DataContentType::PositionDeletes, - file_format: DataFileFormat::Parquet, - file_size_in_bytes: 0, - partition_spec_id: 0, - equality_ids: None, - referenced_data_file: None, - content_offset: None, - content_size_in_bytes: None, - record_count: None, - key_metadata: None, - } + FileScanTaskDeleteFile::builder() + .with_file_path(path.to_string()) + .with_file_type(DataContentType::PositionDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_file_size_in_bytes(0) + .with_partition_spec_id(0) + .build() + .unwrap() + } + + // The sized copy differs from the original in its size alone. Every optional field is set, so + // one the copy failed to forward would show up as a difference. + #[test] + fn sized_delete_file_keeps_every_other_field() { + let original = FileScanTaskDeleteFile::builder() + .with_file_path("s3://bucket/eq-delete.parquet".to_string()) + .with_file_type(DataContentType::EqualityDeletes) + .with_file_format(DataFileFormat::Parquet) + .with_file_size_in_bytes(0) + .with_partition_spec_id(3) + .with_equality_ids(Some(vec![1, 4])) + .with_referenced_data_file(Some("s3://bucket/data.parquet".to_string())) + .with_content_offset(Some(4)) + .with_content_size_in_bytes(Some(40)) + .with_record_count(Some(7)) + .with_key_metadata(Some(Box::from(&b"key"[..]))) + .build() + .unwrap(); + + let sized = IcebergScanExec::sized_delete_file(&original, 1234).unwrap(); + + assert_eq!(sized.file_size_in_bytes(), 1234); + assert_eq!( + IcebergScanExec::sized_delete_file(&sized, 0).unwrap(), + original + ); } // A delete file we cannot stat must fail the scan, not be silently read with a missing/0 size. @@ -720,7 +765,7 @@ mod tests { result.is_err(), "expected an error when a delete file cannot be statted" ); - assert_eq!(tasks[0].deletes()[0].file_size_in_bytes, 0); + assert_eq!(tasks[0].deletes()[0].file_size_in_bytes(), 0); } // The real on-disk size is filled in from the FileIO, replacing the 0 placeholder. @@ -738,7 +783,10 @@ mod tests { .await .unwrap(); - assert_eq!(tasks[0].deletes()[0].file_size_in_bytes, bytes.len() as u64); + assert_eq!( + tasks[0].deletes()[0].file_size_in_bytes(), + bytes.len() as u64 + ); } // A present-but-undersized delete file (0-byte object, truncated write, or a HEAD with no @@ -757,7 +805,7 @@ mod tests { result.is_err(), "expected an error when a delete file is below the Parquet footer minimum" ); - assert_eq!(tasks[0].deletes()[0].file_size_in_bytes, 0); + assert_eq!(tasks[0].deletes()[0].file_size_in_bytes(), 0); } // No deletes means no stats and no error. diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index b88b477931d..0ca5a28169c 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4566,34 +4566,58 @@ fn parse_file_scan_tasks_from_common( file_path )) })?; + // iceberg-rust holds the two deletion-vector coordinates as `u64` too. + let content_offset = + del.content_offset + .map(u64::try_from) + .transpose() + .map_err(|_| { + GeneralError(format!( + "Delete file '{}' has a negative content offset", + file_path + )) + })?; + let content_size_in_bytes = del + .content_size_in_bytes + .map(u64::try_from) + .transpose() + .map_err(|_| { + GeneralError(format!( + "Delete file '{}' has a negative content size", + file_path + )) + })?; - Ok(iceberg::scan::FileScanTaskDeleteFile { + iceberg::scan::FileScanTaskDeleteFile::builder() // Passed RAW, like `data_file_path` below (same exact-string delete-matching // constraint -- see there). - file_path, - file_type, - file_format, + .with_file_path(file_path) + .with_file_type(file_type) + .with_file_format(file_format) // Not serialized; filled in by IcebergScanExec::fill_delete_file_sizes. - file_size_in_bytes: 0, - partition_spec_id: del.partition_spec_id, - equality_ids: if del.equality_ids.is_empty() { + .with_file_size_in_bytes(0) + .with_partition_spec_id(del.partition_spec_id) + .with_equality_ids(if del.equality_ids.is_empty() { None } else { Some(del.equality_ids.clone()) - }, + }) // Deletion-vector coordinates, which the serde sets only when file_format is // PUFFIN. referenced_data_file names the data file the vector applies to; the other // two locate the deletion-vector-v1 blob in its Puffin file. file_format above is // the discriminator, since Iceberg also populates referencedDataFile on // file-scoped Parquet position deletes. - referenced_data_file: del.referenced_data_file.clone(), - content_offset: del.content_offset, - content_size_in_bytes: del.content_size_in_bytes, - record_count, + .with_referenced_data_file(del.referenced_data_file.clone()) + .with_content_offset(content_offset) + .with_content_size_in_bytes(content_size_in_bytes) + .with_record_count(record_count) // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by // iceberg-rust with no KMS unwrap. None for unencrypted delete files. - key_metadata: del.key_metadata.clone().map(Vec::into_boxed_slice), - }) + .with_key_metadata(del.key_metadata.clone().map(Vec::into_boxed_slice)) + // Rejects a deletion vector that lacks its referenced data file, either blob + // coordinate or its record count. + .build() + .map_err(|e| GeneralError(format!("Invalid Iceberg delete file: {e}"))) }) .collect::, ExecutionError>>()?; @@ -7852,6 +7876,69 @@ mod tests { assert_eq!(fields[1].name, "category"); } + /// Delete files go through iceberg-rust's builder, which rejects a deletion vector that lacks + /// its referenced data file, a blob coordinate or its record count. iceberg-rust holds the + /// coordinates as `u64`, so a negative one is rejected before that. + #[test] + fn test_delete_file_pool_validates_deletion_vectors() { + let deletion_vector = spark_operator::IcebergDeleteFile { + file_path_idx: 0, + content_type: "POSITION_DELETES".to_string(), + file_format: "PUFFIN".to_string(), + referenced_data_file: Some("file:///tmp/data.parquet".to_string()), + content_offset: Some(4), + content_size_in_bytes: Some(40), + record_count: Some(2), + ..Default::default() + }; + let parse = |delete: spark_operator::IcebergDeleteFile| { + let proto_common = spark_operator::IcebergScanCommon { + delete_file_path_pool: vec!["file:///tmp/dv.puffin".to_string()], + delete_file_pool: vec![delete], + ..Default::default() + }; + parse_file_scan_tasks_from_common(&proto_common, &[]).map(|_| ()) + }; + + assert!(parse(deletion_vector.clone()).is_ok()); + for (delete, expected) in [ + ( + spark_operator::IcebergDeleteFile { + content_offset: Some(-1), + ..deletion_vector.clone() + }, + "negative content offset", + ), + ( + spark_operator::IcebergDeleteFile { + content_size_in_bytes: Some(-1), + ..deletion_vector.clone() + }, + "negative content size", + ), + ( + spark_operator::IcebergDeleteFile { + referenced_data_file: None, + ..deletion_vector.clone() + }, + "missing referenced_data_file", + ), + ( + spark_operator::IcebergDeleteFile { + record_count: None, + ..deletion_vector.clone() + }, + "missing record_count", + ), + ] { + let error = parse(delete).expect_err(expected).to_string(); + assert!( + error.contains(expected), + "expected '{expected}' in: {error}" + ); + } + } + #[test] fn test_unified_partition_type_tolerates_unparseable_spec() { // A spec whose transform iceberg-rust cannot deserialize still has to yield a spec_id for diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala index 96626631393..bde1f78cbf2 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -812,46 +812,6 @@ object IcebergReflection extends Logging { .toSeq } - /** - * The partition fields of `spec` whose source column is a `float` or `double`, as (partition - * field name, Iceberg type name). Only the identity transform applies to those types, so such a - * field holds the column's own values. A `void` field is skipped: it only ever holds null, and - * its source column may no longer exist. Each field is resolved through its own `sourceId` - * rather than by position in `partitionType()`. Throws on reflection failure, or when a field - * that is not `void` has no source column, so the caller can fail closed. - */ - def floatingPointPartitionFields(spec: Any): Seq[(String, String)] = { - import scala.jdk.CollectionConverters._ - val schema = getMethod(spec.getClass, "schema").invoke(spec) - val findField = getMethod(schema.getClass, "findField", classOf[Int]) - getMethod(spec.getClass, "fields") - .invoke(spec) - .asInstanceOf[java.util.List[_]] - .asScala - .flatMap { partitionField => - val transform = - getMethod(partitionField.getClass, "transform").invoke(partitionField).toString - if (transform == "void") { - None - } else { - val name = - getMethod(partitionField.getClass, "name").invoke(partitionField).asInstanceOf[String] - val sourceId = - getMethod(partitionField.getClass, "sourceId") - .invoke(partitionField) - .asInstanceOf[Int] - val source = findField.invoke(schema, sourceId.asInstanceOf[Object]) - if (source == null) { - throw new IllegalStateException( - s"partition field $name has no source column with id $sourceId") - } - val sourceType = getMethod(source.getClass, "type").invoke(source).toString - if (sourceType == "float" || sourceType == "double") Some(name -> sourceType) else None - } - } - .toSeq - } - /** * The names of the `void` partition fields of `spec` whose source column is no longer in the * spec's schema, when `spec` also has a field that is not `void`. A format-version-1 table diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala index 34c07ff8dd5..2b5cc70f751 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala @@ -398,7 +398,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { requireSupportedFormatVersion, requireNoMetadataColumns, requireSupportedColumnTypes, - requireNoFloatingPointPartitionField, requireNoVoidFieldWithDroppedSource, requireNoEncryptionPrefix, requireNoBloomFilterColumnsEnabled, @@ -514,29 +513,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { } } - // iceberg-rust holds a float partition value as an `OrderedFloat`, whose equality treats -0.0 - // and 0.0 as one value, and its fanout and clustered writers group rows by that equality. - // iceberg-java keeps the two apart, so the native writer would file both under whichever - // arrived first, and a read that prunes on the other value would lose rows (#6138). Remove this - // rule once the iceberg-rust pin carries a fix for apache/iceberg-rust#3325; #5643 tracks it. - private val requireNoFloatingPointPartitionField: TriggerRule = ctx => - IcebergReflection - .getOutputSpecIdFromSparkWrite(ctx.sparkWrite) - .flatMap(IcebergReflection.getPartitionSpecById(ctx.table, _)) match { - case None => Some("could not resolve the output partition spec for type checking") - case Some(spec) => - try { - IcebergReflection.floatingPointPartitionFields(spec).headOption.map { - case (name, typeName) => - s"partition field $name has Iceberg type $typeName, and the native writer does " + - "not keep -0.0 and 0.0 partitions apart" - } - } catch { - case e: Exception => - Some(s"could not inspect the output partition spec: ${e.getMessage}") - } - } - // A format-version-1 spec keeps a dropped partition field as a `void` transform, and its source // column can be dropped afterwards. iceberg-java cannot write through a spec that mixes such a // field with a live one, and the native writer fails resolving the spec's partition type, so diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 36a7a235246..db13435a00c 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -2172,51 +2172,73 @@ class CometIcebergWriteActionSuite } } - // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compares float partition - // values with an equality that treats -0.0 and 0.0 as one value, so its writers filed both under - // whichever arrived first, and a read that pruned on the other value lost rows. iceberg-java - // keeps the two apart, so these writes fall back until iceberg-rust does too. + // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compared float partition + // values with an equality that treated -0.0 and 0.0 as one value, so its writers filed both + // under whichever arrived first, and a read that pruned on the other value lost rows. Since + // apache/iceberg-rust#3327 it keeps them apart, as iceberg-java does. One shuffle partition puts + // every row in one task, so the zeros meet in one writer. The fanout writer gets them + // interleaved. The clustered writer gets one row per partition: Spark's sort ties -0.0 with + // 0.0, so a zero that repeats could come back unclustered, which iceberg-java rejects too. test("signed-zero float and double identity partitions match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => - Seq("zero_native", "zero_jvm").foreach { table => - spark.sql(s""" - CREATE TABLE $catalog.$ns.$table (id INT, f FLOAT, d DOUBLE) - USING iceberg PARTITIONED BY (f, d) - """) - } - val values = - "(1, CAST('-0.0' AS FLOAT), 0.0D), " + - "(2, CAST(0.0 AS FLOAT), 0.0D), " + - "(3, CAST(0.0 AS FLOAT), CAST('-0.0' AS DOUBLE))" - - assertNativeWriteDoesNotEngage("zero_native", Seq(1, 2, 3)) { - spark.sql(s"INSERT INTO $catalog.$ns.zero_native VALUES $values") - } - spark.sql(s"INSERT INTO $catalog.$ns.zero_jvm VALUES $values") + val (negF, posF) = ("CAST('-0.0' AS FLOAT)", "CAST(0.0 AS FLOAT)") + val (negD, posD) = ("CAST('-0.0' AS DOUBLE)", "0.0D") + val onePerPartition = Seq((negF, posD), (posF, posD), (posF, negD), (negF, negD)) + val interleaved = onePerPartition ++ onePerPartition.reverse + Seq("true" -> interleaved, "false" -> onePerPartition).foreach { case (fanout, rows) => + val (nativeTable, jvmTable) = (s"zero_native_$fanout", s"zero_jvm_$fanout") + Seq(nativeTable, jvmTable).foreach { table => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table (id INT, f FLOAT, d DOUBLE) + USING iceberg PARTITIONED BY (f, d) + TBLPROPERTIES ( + 'write.spark.fanout.enabled'='$fanout', 'write.distribution-mode'='hash') + """) + } + val ids = rows.indices.map(_ + 1) + val values = rows.zip(ids).map { case ((f, d), id) => s"($id, $f, $d)" }.mkString(", ") + withSQLConf(SQLConf.SHUFFLE_PARTITIONS.key -> "1") { + assertNativeWriteEngages(nativeTable, ids) { + spark.sql(s"INSERT INTO $catalog.$ns.$nativeTable VALUES $values") + } + spark.sql(s"INSERT INTO $catalog.$ns.$jvmTable VALUES $values") + } - val dirs = partitionDirs(warehouseDir, "zero_native") - assert(dirs == partitionDirs(warehouseDir, "zero_jvm"), s"native enabled: $dirs") - assert(dirs == Set("f=-0.0/d=0.0", "f=0.0/d=0.0", "f=0.0/d=-0.0")) + val dirs = partitionDirs(warehouseDir, nativeTable) + assert(dirs == partitionDirs(warehouseDir, jvmTable), s"fanout=$fanout: $dirs") + assert(dirs == Set("f=-0.0/d=0.0", "f=0.0/d=0.0", "f=0.0/d=-0.0", "f=-0.0/d=-0.0")) - // Iceberg prunes on the partition value, so a merged partition changes what a read returns. - def idsWhereDIsZero(table: String): Seq[Int] = - spark - .sql(s"SELECT id FROM $catalog.$ns.$table WHERE d = 0.0D") + // The committed partition values, not just their spelling in the path. + def partitionValues(table: String): Seq[String] = spark + .sql(s"SELECT CAST(partition AS STRING) FROM $catalog.$ns.$table.files") .collect() - .map(_.getInt(0)) - .sorted + .map(_.getString(0)) .toSeq - assert(idsWhereDIsZero("zero_native") == idsWhereDIsZero("zero_jvm")) + .sorted + assert( + partitionValues(nativeTable) == partitionValues(jvmTable), + s"fanout=$fanout: ${partitionValues(nativeTable)}") + + // Iceberg prunes on the partition value, so a merged partition changes what a read + // returns. + Seq(s"f = $negF", s"f = $posF", s"d = $negD", s"d = $posD").foreach { predicate => + def matching(table: String): Seq[Int] = + spark + .sql(s"SELECT id FROM $catalog.$ns.$table WHERE $predicate") + .collect() + .map(_.getInt(0)) + .sorted + .toSeq + assert(matching(nativeTable) == matching(jvmTable), s"fanout=$fanout, $predicate") + } + } } } // iceberg-java renders a `float` or `double` partition value with `Float.toString` / // `Double.toString`. Rust's `Display` spelled `Double.MAX_VALUE` as 309 digits instead, past the - // 255-byte limit on one path component (apache/datafusion-comet#5836). Float and double identity - // partitions now fall back (#6138), so this checks that the fallback keeps iceberg-java's - // layout; the unit tests in `iceberg_partition_path.rs` cover how the native writer renders - // these values. + // 255-byte limit on one path component (apache/datafusion-comet#5836). test("float and double partition paths match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => @@ -2234,7 +2256,7 @@ class CometIcebergWriteActionSuite "(3, CAST(3.4028235E38 AS FLOAT), CAST(4.9E-324 AS DOUBLE)), " + "(4, CAST(0.001 AS FLOAT), CAST(1.0E20 AS DOUBLE))" - assertNativeWriteDoesNotEngage("float_path_native", Seq(1, 2, 3, 4)) { + assertNativeWriteEngages("float_path_native", Seq(1, 2, 3, 4)) { spark.sql(s"INSERT INTO $catalog.$ns.float_path_native VALUES $values") } spark.sql(s"INSERT INTO $catalog.$ns.float_path_jvm VALUES $values") diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala index 922b5d6073f..4e0cf1eb29e 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala @@ -1500,95 +1500,25 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes } } - test("fall-back: identity partition on a float or double column") { - // iceberg-rust groups float partition values with an equality that treats -0.0 and 0.0 as one - // value, where iceberg-java keeps them apart (#6138). + test("Compatible when partitioned by identity on a float or double column") { + // iceberg-rust grouped float partition values with an equality that treated -0.0 and 0.0 as + // one value, so these writes fell back (#6138) until apache/iceberg-rust#3327 kept the two + // apart as iceberg-java does. A nested source field resolves the same way. withDetectionCatalog { _ => - Seq("float" -> "FLOAT", "double" -> "DOUBLE").foreach { case (typeName, sqlType) => - val table = s"part_$typeName" - spark.sql(s""" - CREATE TABLE $catalog.$ns.$table (id INT, v $sqlType) - USING iceberg PARTITIONED BY (v) - """) - val writeExec = captureWriteExec(table, allowWriteFailure = false) { - spark.sql(s"INSERT INTO $catalog.$ns.$table VALUES (1, CAST(1.5 AS $sqlType))") - } - assertUnsupportedContains(writeExec, table, "partition field v", typeName, "-0.0") - } - } - } - - test("fall-back: identity partition on a nested double field") { - // The source of a partition field can be nested inside a struct. `Schema.findField` resolves - // a nested id too, so the rule must not fail open for it. - withDetectionCatalog { _ => - spark.sql(s""" - CREATE TABLE $catalog.$ns.part_nested (id INT, s STRUCT) - USING iceberg PARTITIONED BY (s.v) - """) - val writeExec = captureWriteExec("part_nested", allowWriteFailure = false) { - spark.sql(s"INSERT INTO $catalog.$ns.part_nested VALUES (1, named_struct('v', 1.5D))") - } - assertUnsupportedContains(writeExec, "part_nested", "partition field s.v", "double", "-0.0") - } - } - - test("fall-back: double identity partition beside a dropped partition field") { - // A format-version-1 spec keeps a dropped partition field as a `void` transform, and that - // field's source column can be dropped afterwards. The surviving double field must still be - // found, whatever the dropped one does to the spec's partition type. - withDetectionCatalog { dir => - createTable( - dir, - "part_dropped", - partitionSpec = "PARTITIONED BY (region, amount)", - properties = Some("'format-version'='1'")) - // Loaded afresh for each change: the insert in between commits through another handle. - def table: org.apache.iceberg.Table = - loadIcebergTable(spark, catalog, ns, "part_dropped") - .asInstanceOf[org.apache.iceberg.Table] - table.updateSpec().removeField("region").commit() - spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped") - assertUnsupportedContains("part_dropped", "partition field amount", "double", "-0.0") - - // Iceberg before 1.11 cannot plan a write once the `void` field's source column is gone. - // On 1.11 iceberg-java plans it but cannot build a partition key for a spec that mixes that - // field with a live one, so the write itself fails on either path; only the gate's decision - // is checked. - if (icebergVersionAtLeast(1, 11)) { - table.updateSchema().deleteColumn("region").commit() - spark.sql(s"REFRESH TABLE $catalog.$ns.part_dropped") - val writeExec = captureWriteExec("part_dropped", allowWriteFailure = true) { - spark.sql(s"INSERT INTO $catalog.$ns.part_dropped VALUES (2, 2.0)") - } - assertUnsupportedContains( - writeExec, - "part_dropped", - "partition field amount", - "double", - "-0.0") + Seq( + "part_float" -> ("v FLOAT", "v", "CAST(1.5 AS FLOAT)"), + "part_double" -> ("v DOUBLE", "v", "1.5D"), + "part_nested" -> ("s STRUCT", "s.v", "named_struct('v', 1.5D)")).foreach { + case (table, (column, partitionField, value)) => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table (id INT, $column) + USING iceberg PARTITIONED BY ($partitionField) + """) + assertSupportLevelIs[Compatible](table, values = s"(1, $value)") } } } - test("Compatible when a dropped double partition field remains as void") { - // The `void` field only ever holds null, so there are no signed zeros to keep apart. - withDetectionCatalog { dir => - createTable( - dir, - "part_void", - partitionSpec = "PARTITIONED BY (amount)", - properties = Some("'format-version'='1'")) - loadIcebergTable(spark, catalog, ns, "part_void") - .asInstanceOf[org.apache.iceberg.Table] - .updateSpec() - .removeField("amount") - .commit() - spark.sql(s"REFRESH TABLE $catalog.$ns.part_void") - assertSupportLevelIs[Compatible]("part_void") - } - } - // A format-version-1 spec keeps a dropped partition field as a `void` transform, and the // field's source column can be dropped afterwards. Neither writer can write through a spec that // mixes such a field with a live one: iceberg-java fails building the partition key, and the @@ -1788,10 +1718,11 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes private def assertSupportLevelIs[T <: SupportLevel: scala.reflect.ClassTag]( tableName: String, - allowWriteFailure: Boolean = false): Unit = { + allowWriteFailure: Boolean = false, + values: String = "(1, 'us', 1.0)"): Unit = { val expected = scala.reflect.classTag[T].runtimeClass val plan = captureWritePlan(tableName, allowWriteFailure) { - spark.sql(s"INSERT INTO $catalog.$ns.$tableName VALUES (1, 'us', 1.0)") + spark.sql(s"INSERT INTO $catalog.$ns.$tableName VALUES $values") } findWriteExec(plan) match { case Some(writeExec) => From 1036ae5d6f82f83c15e28408c3ad19d3e6fd3c8f Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 10 Oct 2026 07:46:44 -0600 Subject: [PATCH 2/5] test: cover NaN and same-typed decimal identity partitions The lifted gate also admits NaN partition values, so the signed-zero test writes NaN and -NaN through both writers too. The bump changes how iceberg-rust writes a named Avro type that two partition fields share, so a new test writes two decimal identity partitions of the same type, in both encodings, and compares them with iceberg-java. The detection test now uses plain tuples. Scala 2.12 flagged the adapted argument list as an error under -Pstrict-warnings. --- .../comet/CometIcebergWriteActionSuite.scala | 106 +++++++++++++----- .../CometIcebergWriteDetectionSuite.scala | 8 +- 2 files changed, 79 insertions(+), 35 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index db13435a00c..9c3e3883783 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -2175,18 +2175,21 @@ class CometIcebergWriteActionSuite // https://github.com/apache/datafusion-comet/issues/6138. iceberg-rust compared float partition // values with an equality that treated -0.0 and 0.0 as one value, so its writers filed both // under whichever arrived first, and a read that pruned on the other value lost rows. Since - // apache/iceberg-rust#3327 it keeps them apart, as iceberg-java does. One shuffle partition puts - // every row in one task, so the zeros meet in one writer. The fanout writer gets them - // interleaved. The clustered writer gets one row per partition: Spark's sort ties -0.0 with - // 0.0, so a zero that repeats could come back unclustered, which iceberg-java rejects too. - test("signed-zero float and double identity partitions match iceberg-java") { + // apache/iceberg-rust#3327 it keeps them apart and puts every NaN in one partition, as + // iceberg-java does. One shuffle partition puts every row in one task, so the values meet in one + // writer. The fanout writer gets them interleaved. The clustered writer gets each partition's + // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats could come back + // unclustered, which iceberg-java rejects too. + test("signed-zero and NaN float and double identity partitions match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => - val (negF, posF) = ("CAST('-0.0' AS FLOAT)", "CAST(0.0 AS FLOAT)") - val (negD, posD) = ("CAST('-0.0' AS DOUBLE)", "0.0D") - val onePerPartition = Seq((negF, posD), (posF, posD), (posF, negD), (negF, negD)) - val interleaved = onePerPartition ++ onePerPartition.reverse - Seq("true" -> interleaved, "false" -> onePerPartition).foreach { case (fanout, rows) => + val (negF, posF, nanF) = + ("CAST('-0.0' AS FLOAT)", "CAST(0.0 AS FLOAT)", "CAST('NaN' AS FLOAT)") + val (negD, posD, nanD) = ("CAST('-0.0' AS DOUBLE)", "0.0D", "CAST('NaN' AS DOUBLE)") + val zeros = Seq((negF, posD), (posF, posD), (posF, negD), (negF, negD)) + val nans = Seq((nanF, nanD), (s"-$nanF", s"-$nanD")) + val interleaved = zeros ++ nans ++ zeros.reverse ++ nans + Seq("true" -> interleaved, "false" -> (zeros ++ nans)).foreach { case (fanout, rows) => val (nativeTable, jvmTable) = (s"zero_native_$fanout", s"zero_jvm_$fanout") Seq(nativeTable, jvmTable).foreach { table => spark.sql(s""" @@ -2207,31 +2210,30 @@ class CometIcebergWriteActionSuite val dirs = partitionDirs(warehouseDir, nativeTable) assert(dirs == partitionDirs(warehouseDir, jvmTable), s"fanout=$fanout: $dirs") - assert(dirs == Set("f=-0.0/d=0.0", "f=0.0/d=0.0", "f=0.0/d=-0.0", "f=-0.0/d=-0.0")) - - // The committed partition values, not just their spelling in the path. - def partitionValues(table: String): Seq[String] = spark - .sql(s"SELECT CAST(partition AS STRING) FROM $catalog.$ns.$table.files") - .collect() - .map(_.getString(0)) - .toSeq - .sorted assert( - partitionValues(nativeTable) == partitionValues(jvmTable), - s"fanout=$fanout: ${partitionValues(nativeTable)}") + dirs == Set( + "f=-0.0/d=0.0", + "f=0.0/d=0.0", + "f=0.0/d=-0.0", + "f=-0.0/d=-0.0", + "f=NaN/d=NaN")) + assert( + committedPartitions(nativeTable) == committedPartitions(jvmTable), + s"fanout=$fanout: ${committedPartitions(nativeTable)}") // Iceberg prunes on the partition value, so a merged partition changes what a read // returns. - Seq(s"f = $negF", s"f = $posF", s"d = $negD", s"d = $posD").foreach { predicate => - def matching(table: String): Seq[Int] = - spark - .sql(s"SELECT id FROM $catalog.$ns.$table WHERE $predicate") - .collect() - .map(_.getInt(0)) - .sorted - .toSeq - assert(matching(nativeTable) == matching(jvmTable), s"fanout=$fanout, $predicate") - } + Seq(s"f = $negF", s"f = $posF", s"d = $negD", s"d = $posD", s"d = $nanD", "isnan(f)") + .foreach { predicate => + def matching(table: String): Seq[Int] = + spark + .sql(s"SELECT id FROM $catalog.$ns.$table WHERE $predicate") + .collect() + .map(_.getInt(0)) + .sorted + .toSeq + assert(matching(nativeTable) == matching(jvmTable), s"fanout=$fanout, $predicate") + } } } } @@ -2272,6 +2274,39 @@ class CometIcebergWriteActionSuite } } + // Two partition fields of the same decimal type share one named Avro type, `decimal_10_2`. + // iceberg-rust defined it once per field, which the Avro specification forbids, until it moved + // to apache-avro 0.22 (apache/iceberg-rust#3354). iceberg-java reads each native task's manifest + // back, so this checks it still can. `decimal(38, 2)` covers the fixed-length encoding. + test("two decimal identity partitions of the same type match iceberg-java") { + assumeNativeAcceleration() + withIcebergCatalog { warehouseDir => + Seq("10, 2", "38, 2").foreach { decimalType => + val suffix = decimalType.replace(", ", "_") + val (nativeTable, jvmTable) = (s"decimal_native_$suffix", s"decimal_jvm_$suffix") + Seq(nativeTable, jvmTable).foreach { table => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table ( + id INT, d1 DECIMAL($decimalType), d2 DECIMAL($decimalType)) + USING iceberg PARTITIONED BY (d1, d2) + """) + } + val values = "(1, 123.45, -6.78), (2, 0.00, 1.10)" + assertNativeWriteEngages(nativeTable, Seq(1, 2)) { + spark.sql(s"INSERT INTO $catalog.$ns.$nativeTable VALUES $values") + } + spark.sql(s"INSERT INTO $catalog.$ns.$jvmTable VALUES $values") + + val dirs = partitionDirs(warehouseDir, nativeTable) + assert(dirs == partitionDirs(warehouseDir, jvmTable), s"decimal($decimalType): $dirs") + assert(dirs == Set("d1=123.45/d2=-6.78", "d1=0.00/d2=1.10")) + assert( + committedPartitions(nativeTable) == committedPartitions(jvmTable), + s"decimal($decimalType): ${committedPartitions(nativeTable)}") + } + } + } + // https://github.com/apache/datafusion-comet/issues/6145. iceberg-rust splits the calendar with // `chrono`, which stops at year 262142, while a Spark date reaches year 5881580 and a timestamp // year 294247. Past that, rendering a `days` or identity-date partition directory panicked the @@ -4083,6 +4118,15 @@ class CometIcebergWriteActionSuite * location. Comparing this set between a natively-written table and a JVM-written twin pins the * on-disk layout against iceberg-java's `PartitionSpec#partitionToPath`. */ + /** Each data file's committed partition value and record count, sorted. */ + private def committedPartitions(tableName: String): Seq[String] = + spark + .sql(s"SELECT CAST(partition AS STRING), record_count FROM $catalog.$ns.$tableName.files") + .collect() + .map(_.toString) + .toSeq + .sorted + private def partitionDirs(warehouseDir: File, tableName: String): Set[String] = { val location = warehouseDir.toURI.toString.stripSuffix("/") spark diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala index 4e0cf1eb29e..9c8e6b96ab7 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala @@ -1506,10 +1506,10 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes // apart as iceberg-java does. A nested source field resolves the same way. withDetectionCatalog { _ => Seq( - "part_float" -> ("v FLOAT", "v", "CAST(1.5 AS FLOAT)"), - "part_double" -> ("v DOUBLE", "v", "1.5D"), - "part_nested" -> ("s STRUCT", "s.v", "named_struct('v', 1.5D)")).foreach { - case (table, (column, partitionField, value)) => + ("part_float", "v FLOAT", "v", "CAST(1.5 AS FLOAT)"), + ("part_double", "v DOUBLE", "v", "1.5D"), + ("part_nested", "s STRUCT", "s.v", "named_struct('v', 1.5D)")).foreach { + case (table, column, partitionField, value) => spark.sql(s""" CREATE TABLE $catalog.$ns.$table (id INT, $column) USING iceberg PARTITIONED BY ($partitionField) From 3ffae5ad24946525e81e8a4c8b13cab52348b9e3 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 10 Oct 2026 08:02:17 -0600 Subject: [PATCH 3/5] refactor: share the delete-file count conversion and trim the new tests - One closure converts a delete file's record count and blob coordinates from the proto's int64, instead of three copies of the same block. - The planner test keeps the cases Comet's own conversion decides, and drops the two that only exercised iceberg-rust's validation messages. - `committedPartitions` sits after `partitionDirs`, replaces the far-date test's local copy of the same query, and is read once per assertion. - The detection test uses the suite's `createTable` table, so `assertSupportLevelIs` keeps its original signature. --- native/core/src/execution/planner.rs | 62 ++++--------------- .../comet/CometIcebergWriteActionSuite.scala | 41 +++++------- .../CometIcebergWriteDetectionSuite.scala | 24 +++---- 3 files changed, 36 insertions(+), 91 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 0ca5a28169c..009431ed792 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4554,39 +4554,18 @@ fn parse_file_scan_tasks_from_common( })? .clone(); + // iceberg-rust holds these counts and offsets as `u64`, so a negative one is rejected + // here, where the proto's `int64` becomes iceberg-rust's type. + let non_negative = |value: Option, what: &str| { + value.map(u64::try_from).transpose().map_err(|_| { + GeneralError(format!("Delete file '{file_path}' has a negative {what}")) + }) + }; // Required for deletion vectors: iceberg-rust checks it against the number of // positions decoded from the blob and errors if it is absent. - let record_count = del - .record_count - .map(u64::try_from) - .transpose() - .map_err(|_| { - GeneralError(format!( - "Delete file '{}' has a negative record count", - file_path - )) - })?; - // iceberg-rust holds the two deletion-vector coordinates as `u64` too. - let content_offset = - del.content_offset - .map(u64::try_from) - .transpose() - .map_err(|_| { - GeneralError(format!( - "Delete file '{}' has a negative content offset", - file_path - )) - })?; - let content_size_in_bytes = del - .content_size_in_bytes - .map(u64::try_from) - .transpose() - .map_err(|_| { - GeneralError(format!( - "Delete file '{}' has a negative content size", - file_path - )) - })?; + let record_count = non_negative(del.record_count, "record count")?; + let content_offset = non_negative(del.content_offset, "content offset")?; + let content_size_in_bytes = non_negative(del.content_size_in_bytes, "content size")?; iceberg::scan::FileScanTaskDeleteFile::builder() // Passed RAW, like `data_file_path` below (same exact-string delete-matching @@ -7876,11 +7855,10 @@ mod tests { assert_eq!(fields[1].name, "category"); } - /// Delete files go through iceberg-rust's builder, which rejects a deletion vector that lacks - /// its referenced data file, a blob coordinate or its record count. iceberg-rust holds the - /// coordinates as `u64`, so a negative one is rejected before that. + /// iceberg-rust holds a deletion vector's blob coordinates as `u64`, so the planner rejects a + /// negative one from the proto rather than build the delete file with it. #[test] - fn test_delete_file_pool_validates_deletion_vectors() { + fn test_delete_file_pool_rejects_negative_coordinates() { let deletion_vector = spark_operator::IcebergDeleteFile { file_path_idx: 0, content_type: "POSITION_DELETES".to_string(), @@ -7916,20 +7894,6 @@ mod tests { }, "negative content size", ), - ( - spark_operator::IcebergDeleteFile { - referenced_data_file: None, - ..deletion_vector.clone() - }, - "missing referenced_data_file", - ), - ( - spark_operator::IcebergDeleteFile { - record_count: None, - ..deletion_vector.clone() - }, - "missing record_count", - ), ] { let error = parse(delete).expect_err(expected).to_string(); assert!( diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 9c3e3883783..210008c83bd 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -2217,9 +2217,8 @@ class CometIcebergWriteActionSuite "f=0.0/d=-0.0", "f=-0.0/d=-0.0", "f=NaN/d=NaN")) - assert( - committedPartitions(nativeTable) == committedPartitions(jvmTable), - s"fanout=$fanout: ${committedPartitions(nativeTable)}") + val committed = committedPartitions(nativeTable) + assert(committed == committedPartitions(jvmTable), s"fanout=$fanout: $committed") // Iceberg prunes on the partition value, so a merged partition changes what a read // returns. @@ -2300,9 +2299,8 @@ class CometIcebergWriteActionSuite val dirs = partitionDirs(warehouseDir, nativeTable) assert(dirs == partitionDirs(warehouseDir, jvmTable), s"decimal($decimalType): $dirs") assert(dirs == Set("d1=123.45/d2=-6.78", "d1=0.00/d2=1.10")) - assert( - committedPartitions(nativeTable) == committedPartitions(jvmTable), - s"decimal($decimalType): ${committedPartitions(nativeTable)}") + val committed = committedPartitions(nativeTable) + assert(committed == committedPartitions(jvmTable), s"decimal($decimalType): $committed") } } } @@ -2357,16 +2355,9 @@ class CometIcebergWriteActionSuite s"native: $nativeDirs") // The committed partition values, not just their spelling in the path. - def partitionValues(table: String): Seq[String] = spark - .sql(s"SELECT CAST(partition AS STRING) FROM $catalog.$ns.$table.files") - .collect() - .map(_.getString(0)) - .toSeq - .sorted - assert( - partitionValues(nativeTable) == partitionValues(jvmTable), - s"native: ${partitionValues(nativeTable)}") - assert(!partitionValues(nativeTable).exists(_.contains("null"))) + val committed = committedPartitions(nativeTable) + assert(committed == committedPartitions(jvmTable), s"native: $committed") + assert(!committed.exists(_.contains("null"))) // `java.sql.Date` cannot hold these dates (`collect` overflows rebasing them), so read // them back as `LocalDate` and `Instant`. @@ -4118,15 +4109,6 @@ class CometIcebergWriteActionSuite * location. Comparing this set between a natively-written table and a JVM-written twin pins the * on-disk layout against iceberg-java's `PartitionSpec#partitionToPath`. */ - /** Each data file's committed partition value and record count, sorted. */ - private def committedPartitions(tableName: String): Seq[String] = - spark - .sql(s"SELECT CAST(partition AS STRING), record_count FROM $catalog.$ns.$tableName.files") - .collect() - .map(_.toString) - .toSeq - .sorted - private def partitionDirs(warehouseDir: File, tableName: String): Set[String] = { val location = warehouseDir.toURI.toString.stripSuffix("/") spark @@ -4140,6 +4122,15 @@ class CometIcebergWriteActionSuite .toSet } + /** Each data file's committed partition value and record count, sorted. */ + private def committedPartitions(tableName: String): Seq[String] = + spark + .sql(s"SELECT CAST(partition AS STRING), record_count FROM $catalog.$ns.$tableName.files") + .collect() + .map(_.toString) + .toSeq + .sorted + /** * The table's NaN counts summed over its data files, keyed by field id. Two tables created from * the same DDL get the same ids, so their maps compare directly. diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala index 9c8e6b96ab7..03d74341b86 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala @@ -1500,22 +1500,13 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes } } - test("Compatible when partitioned by identity on a float or double column") { + test("Compatible when partitioned by identity on a double column") { // iceberg-rust grouped float partition values with an equality that treated -0.0 and 0.0 as // one value, so these writes fell back (#6138) until apache/iceberg-rust#3327 kept the two - // apart as iceberg-java does. A nested source field resolves the same way. - withDetectionCatalog { _ => - Seq( - ("part_float", "v FLOAT", "v", "CAST(1.5 AS FLOAT)"), - ("part_double", "v DOUBLE", "v", "1.5D"), - ("part_nested", "s STRUCT", "s.v", "named_struct('v', 1.5D)")).foreach { - case (table, column, partitionField, value) => - spark.sql(s""" - CREATE TABLE $catalog.$ns.$table (id INT, $column) - USING iceberg PARTITIONED BY ($partitionField) - """) - assertSupportLevelIs[Compatible](table, values = s"(1, $value)") - } + // apart as iceberg-java does. + withDetectionCatalog { dir => + createTable(dir, "part_double", partitionSpec = "PARTITIONED BY (amount)") + assertSupportLevelIs[Compatible]("part_double") } } @@ -1718,11 +1709,10 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes private def assertSupportLevelIs[T <: SupportLevel: scala.reflect.ClassTag]( tableName: String, - allowWriteFailure: Boolean = false, - values: String = "(1, 'us', 1.0)"): Unit = { + allowWriteFailure: Boolean = false): Unit = { val expected = scala.reflect.classTag[T].runtimeClass val plan = captureWritePlan(tableName, allowWriteFailure) { - spark.sql(s"INSERT INTO $catalog.$ns.$tableName VALUES $values") + spark.sql(s"INSERT INTO $catalog.$ns.$tableName VALUES (1, 'us', 1.0)") } findWriteExec(plan) match { case Some(writeExec) => From 4eae2874fe160ad53800c089d6d647913a982a72 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Tue, 13 Oct 2026 09:01:03 -0600 Subject: [PATCH 4/5] test: check the clustered writer rejects interleaved signed zeros like iceberg-java --- .../comet/CometIcebergWriteActionSuite.scala | 36 ++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 0ed1c3f24f0..6baeb249523 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -2158,7 +2158,7 @@ class CometIcebergWriteActionSuite // iceberg-java does. One shuffle partition puts every row in one task, so the values meet in one // writer. The fanout writer gets them interleaved. The clustered writer gets each partition's // rows together: Spark's sort ties -0.0 with 0.0, so a zero that repeats could come back - // unclustered, which iceberg-java rejects too. + // unclustered, which both writers reject (see the next test). test("signed-zero and NaN float and double identity partitions match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => @@ -2216,6 +2216,40 @@ class CometIcebergWriteActionSuite } } + // https://github.com/apache/datafusion-comet/issues/6138. A clustered writer rejects a row for a + // partition it has already closed. While iceberg-rust treated -0.0 and 0.0 as one value, the + // native writer accepted these rows as one partition, where iceberg-java rejects them. Spark's + // sort ties the two zeros, so `appendUnclustered` skips it to fix the order the writers see. + test("clustered writer rejects interleaved signed zeros like the JVM writer") { + assumeNativeAcceleration() + withIcebergCatalog { _ => + val (nativeTable, jvmTable) = ("zero_unclustered_native", "zero_unclustered_jvm") + val tables = Seq(nativeTable, jvmTable) + tables.foreach { table => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table (id INT, f FLOAT, d DOUBLE) + USING iceberg PARTITIONED BY (f, d) + """) + } + // The -0.0 partition is revisited after the writer has moved on to 0.0 and closed it. + val (negF, posF) = ("CAST('-0.0' AS FLOAT)", "CAST(0.0 AS FLOAT)") + val rows = spark.sql( + s"SELECT * FROM VALUES (1, $negF, 0.0D), (2, $posF, 0.0D), (3, $negF, 0.0D) " + + "AS t(id, f, d)") + + val (nativeFailure, jvmFailure) = + failedWrites(nativeTable, jvmTable)(appendUnclustered(_, rows)) + + val (_, nativeMessage) = clusteredWriterError(nativeFailure) + assert( + clusteredWriterError(nativeFailure) == clusteredWriterError(jvmFailure), + s"native message differs from the JVM writer's:\n$nativeMessage") + assert(nativeMessage.contains("partition 'f=-0.0/d=0.0'"), nativeMessage) + // Neither write may have committed. + tables.foreach(assertRows(_, Seq.empty)) + } + } + // iceberg-java renders a `float` or `double` partition value with `Float.toString` / // `Double.toString`. Rust's `Display` spelled `Double.MAX_VALUE` as 309 digits instead, past the // 255-byte limit on one path component (apache/datafusion-comet#5836). From f6c69ce4a04da4be6fe568c709c0ed4c483406ab Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Thu, 15 Oct 2026 14:28:28 -0600 Subject: [PATCH 5/5] test: cover a missing or negative deletion vector record count; drop the fixed day difference from the guide --- .../contributor-guide/iceberg-writes.md | 5 ++--- native/core/src/execution/planner.rs | 22 ++++++++++++++++--- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md index a30a3192f2f..ecdddb97954 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -246,9 +246,8 @@ Points where Comet adapts iceberg-rust to match iceberg-java: NULL past `chrono`'s calendar ([#6145](https://github.com/apache/datafusion-comet/issues/6145)). All four of its transforms floor a pre-1970 timestamp that lies exactly 999999 microseconds into a unit, which iceberg-java puts in the unit before - ([#6426](https://github.com/apache/datafusion-comet/issues/6426)), and its `day` puts some - timestamps from the last second before a pre-1970 midnight in the next day. Every other transform - stays on iceberg-rust. + ([#6426](https://github.com/apache/datafusion-comet/issues/6426)). Every other transform stays on + iceberg-rust. - **File names.** `file_name_prefix` embeds the partition id, the task attempt id and the operation id, so a retried or speculative attempt never reuses another attempt's file names. - **Row pacing.** iceberg-java's rolling writer checks the target file size every 1000 rows of the diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index c6008cf481c..c773b2d63c6 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -8216,10 +8216,12 @@ mod tests { assert_eq!(fields[1].name, "category"); } - /// iceberg-rust holds a deletion vector's blob coordinates as `u64`, so the planner rejects a - /// negative one from the proto rather than build the delete file with it. + /// iceberg-rust holds a deletion vector's record count and blob coordinates as `u64`, so the + /// planner rejects a negative one from the proto rather than build the delete file with it. A + /// deletion vector without a record count reaches iceberg-rust's builder, whose error must + /// come back through the `Invalid Iceberg delete file` mapping. #[test] - fn test_delete_file_pool_rejects_negative_coordinates() { + fn test_delete_file_pool_rejects_invalid_deletion_vectors() { let deletion_vector = spark_operator::IcebergDeleteFile { file_path_idx: 0, content_type: "POSITION_DELETES".to_string(), @@ -8255,6 +8257,20 @@ mod tests { }, "negative content size", ), + ( + spark_operator::IcebergDeleteFile { + record_count: Some(-1), + ..deletion_vector.clone() + }, + "negative record count", + ), + ( + spark_operator::IcebergDeleteFile { + record_count: None, + ..deletion_vector.clone() + }, + "is missing record_count", + ), ] { let error = parse(delete).expect_err(expected).to_string(); assert!(