diff --git a/.ai/skills/review-comet-iceberg-write-pr/SKILL.md b/.ai/skills/review-comet-iceberg-write-pr/SKILL.md index 0344f0fa267..2f1c2f6cbdf 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 6f930733b17..a30a3192f2f 100644 --- a/docs/source/contributor-guide/iceberg-writes.md +++ b/docs/source/contributor-guide/iceberg-writes.md @@ -414,8 +414,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 cea6ca42a92..40db742ad0f 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, apart from the ignored settings listed below | | 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.*` — @@ -259,13 +259,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 @@ -407,9 +400,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 7bc355b68eb..ccf797c9845 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", @@ -3384,6 +3390,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" @@ -3583,7 +3595,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", @@ -3627,7 +3639,7 @@ dependencies = [ "serde_json", "serde_repr", "serde_with", - "strum", + "strum 0.27.2", "tokio", "tracing", "typed-builder", @@ -3641,7 +3653,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", @@ -3651,7 +3663,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", @@ -4876,6 +4888,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" @@ -5298,6 +5334,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" @@ -5338,7 +5387,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", @@ -6472,6 +6521,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" @@ -6490,7 +6545,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]] @@ -6499,7 +6563,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", @@ -7494,6 +7570,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 defbf41ae27..a7a29db1e31 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 7b18c7e3bea..9f76ee977df 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -278,14 +278,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() { @@ -342,19 +342,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 @@ -696,19 +718,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. @@ -726,7 +771,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. @@ -744,7 +789,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 @@ -763,7 +811,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 a05de76019d..c6008cf481c 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4660,46 +4660,49 @@ 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 - )) - })?; + 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")?; - 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>>()?; @@ -8213,6 +8216,54 @@ 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. + #[test] + fn test_delete_file_pool_rejects_negative_coordinates() { + 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", + ), + ] { + 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 354ff37e215..9fcafd9e8ef 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 42220ea9b28..de77f82b565 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 @@ -417,7 +417,6 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] { requireSupportedFormatVersion, requireNoMetadataColumns, requireSupportedColumnTypes, - requireNoFloatingPointPartitionField, requireNoVoidFieldWithDroppedSource, requireNoEncryptionPrefix, requireNoBloomFilterColumnsEnabled, @@ -533,29 +532,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 489a5db6689..6baeb249523 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -2151,51 +2151,108 @@ 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. - test("signed-zero float and double identity partitions match iceberg-java") { + // 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 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 both writers reject (see the next test). + test("signed-zero and NaN float and double identity partitions match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => - Seq("zero_native", "zero_jvm").foreach { table => + 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""" + 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, 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", + "f=NaN/d=NaN")) + 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. + 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") + } + } + } + } + + // 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) """) } - 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") + // 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 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 (nativeFailure, jvmFailure) = + failedWrites(nativeTable, jvmTable)(appendUnclustered(_, rows)) - // 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") - .collect() - .map(_.getInt(0)) - .sorted - .toSeq - assert(idsWhereDIsZero("zero_native") == idsWhereDIsZero("zero_jvm")) + 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). 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 => @@ -2213,7 +2270,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") @@ -2229,6 +2286,38 @@ 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")) + val committed = committedPartitions(nativeTable) + assert(committed == committedPartitions(jvmTable), s"decimal($decimalType): $committed") + } + } + } + // 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 @@ -2279,16 +2368,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`. @@ -4101,6 +4183,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 2d0c40357f6..6faeaf046e7 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala @@ -1504,90 +1504,13 @@ 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 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. 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 { _ => - createTable( - "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") - } - } - } - - 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 { _ => - createTable( - "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") + createTable("part_double", partitionSpec = "PARTITIONED BY (amount)") + assertSupportLevelIs[Compatible]("part_double") } }