From d6a25964d36aef9a5ae8f6fd169325989f2d40ef Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sat, 5 Sep 2026 16:59:32 -0600 Subject: [PATCH 1/2] fix: native Iceberg write panics on an evolved partition spec and on a timestamptz partition path Two native Iceberg write panics that crossed the JNI boundary as a CometNativeException instead of surfacing as an error. Fixes #5691. iceberg-java keeps a dropped partition field in a format-version-1 spec as a `void` transform, and `isUnpartitioned` means "every field is void" rather than "no fields" on both the Java and the Rust side. Such a write therefore runs through `UnpartitionedWriter`, which stamps every data file with an empty partition struct, while `ManifestWriter` derives one partition summary per spec field and `zip_eq`s the two. Encode the per-task transport manifest against a field-less spec of the same id so the two agree. Nothing downstream loses information: the JVM rebuilds each `DataFile` against the real output spec, whose `DataFiles.Builder` drops partition data for an unpartitioned spec anyway, so the manifest that reaches storage is unchanged. Also fixes #5693, the same shape one step further along -- once the `void` field's source column has itself been dropped, resolving the manifest's partition type failed with "No column with source column id", which the field-less spec no longer needs to do. Fixes #5694. iceberg-rust renders a `timestamptz` partition value by casting `micros % 1_000_000` to `u32`, so a pre-1970 value with a sub-second part unwraps a `None`. Generate the partition path in Comet instead, mirroring iceberg-java's `PartitionSpec#partitionToPath`. That also closes the two other divergences in the same function: `timestamp` and `timestamptz` were rendered with chrono's `Display` rather than ISO-8601, and `binary`/`fixed` with hex rather than base64. `float` and `double` still delegate to iceberg-rust and are documented as a known divergence. --- .../user-guide/latest/iceberg-writes.md | 6 + native/Cargo.lock | 1 + native/core/Cargo.toml | 1 + .../operators/iceberg_partition_path.rs | 535 ++++++++++++++++++ .../src/execution/operators/iceberg_write.rs | 274 ++++++++- native/core/src/execution/operators/mod.rs | 1 + .../comet/CometIcebergWriteActionSuite.scala | 154 ++++- 7 files changed, 949 insertions(+), 23 deletions(-) create mode 100644 native/core/src/execution/operators/iceberg_partition_path.rs diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index 49f35009202..d4159dae6a6 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -251,6 +251,12 @@ a data file but not what any reader computes from it: differences (iceberg-java checks the target file size every 1000 rows and names files `---`; iceberg-rust checks per batch and uses a process-local counter). +- Partition directory names match iceberg-java's `PartitionSpec.partitionToPath` for every + partition type except `float` and `double`, where the value is rendered with Rust's shortest + representation instead of `Float.toString` / `Double.toString` (`f=1` where iceberg-java writes + `f=1.0`). Distinct partition values still get distinct directories, and no reader parses these + names — files are resolved through committed manifests. Iceberg deprecated float and double + partitioning in 1.3. - Compressed page bytes are implementation-defined: the codec and any explicit level are translated, but parquet-rs and parquet-mr embed different encoder implementations and defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even diff --git a/native/Cargo.lock b/native/Cargo.lock index d3a4b897df2..e8369070804 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -1997,6 +1997,7 @@ dependencies = [ "async-trait", "aws-config", "aws-credential-types", + "base64 0.23.1", "bytes", "comet-contrib-delta", "criterion", diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml index 787d648c4be..ae005365335 100644 --- a/native/core/Cargo.toml +++ b/native/core/Cargo.toml @@ -36,6 +36,7 @@ publish = false [dependencies] arrow = { workspace = true } +base64 = "0.23.0" bytes = { workspace = true } parquet = { workspace = true, default-features = false, features = ["experimental", "arrow", "snap", "lz4", "zstd", "flate2-zlib-rs"] } futures = { workspace = true } diff --git a/native/core/src/execution/operators/iceberg_partition_path.rs b/native/core/src/execution/operators/iceberg_partition_path.rs new file mode 100644 index 00000000000..ea609a59b4e --- /dev/null +++ b/native/core/src/execution/operators/iceberg_partition_path.rs @@ -0,0 +1,535 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Data-file location generation for the native Iceberg writer. +//! +//! [`CometLocationGenerator`] stands in for iceberg-rust's `DefaultLocationGenerator`. It lays out +//! files identically (`{data_location}/{partition_path}/{file_name}`), but renders the partition +//! path itself so the directory names match iceberg-java's `PartitionSpec#partitionToPath` rather +//! than iceberg-rust's `PartitionSpec::partition_to_path`, which diverges from it -- and, for a +//! pre-1970 `timestamptz` value, panics. + +use std::sync::Arc; + +use base64::engine::general_purpose::STANDARD as BASE64; +use base64::Engine; +use iceberg::spec::{ + Literal, PartitionKey, PartitionSpec, PrimitiveLiteral, PrimitiveType, SchemaRef, StructType, + Transform, Type, +}; +use iceberg::writer::file_writer::location_generator::LocationGenerator; +use itertools::Itertools; +use url::form_urlencoded; + +const SECONDS_PER_DAY: i64 = 86_400; + +/// `LocationGenerator` for Comet's native Iceberg writer. +/// +/// `partition_type` is the write's partition spec resolved against its schema, computed once when +/// the writer stack is built. `LocationGenerator::generate_location` cannot fail, so resolving it up +/// front turns a spec that cannot be resolved into an error at task start instead of a panic on +/// every file (iceberg-rust's `PartitionSpec::partition_to_path` unwraps the same call inline). +#[derive(Clone, Debug)] +pub struct CometLocationGenerator { + data_location: String, + partition_type: Arc, +} + +impl CometLocationGenerator { + /// `data_location` is used verbatim as the parent directory, matching + /// `DefaultLocationGenerator::with_data_location`. + pub fn try_new( + data_location: String, + partition_spec: &PartitionSpec, + schema: &SchemaRef, + ) -> Result { + // An unpartitioned spec never contributes a partition directory (see + // `generate_location`), so its partition type is never read. Skipping the resolution + // matters: a V1 spec keeps a dropped partition field as a `void` transform, and that + // field's source column may since have been dropped from the schema -- which + // `PartitionSpec::partition_type` rejects even though nothing needs the answer. + let partition_type = if partition_spec.is_unpartitioned() { + StructType::new(vec![]) + } else { + partition_spec.partition_type(schema)? + }; + Ok(Self { + data_location, + partition_type: Arc::new(partition_type), + }) + } + + /// Mirrors iceberg-java's `PartitionSpec#partitionToPath`: `name=value` pairs joined by `/`, + /// with both halves form-urlencoded. `form_urlencoded` leaves exactly the byte set Java's + /// `URLEncoder.encode(s, UTF_8)` leaves (`A-Za-z0-9`, `*`, `-`, `.`, `_`), maps space to `+`, + /// and percent-encodes the rest with uppercase hex, so the two agree byte for byte. + fn partition_to_path(&self, key: &PartitionKey) -> String { + let fields = self.partition_type.fields(); + key.spec() + .fields() + .iter() + .enumerate() + .map(|(index, field)| { + // Indexed rather than zipped so a spec/partition-type/value length disagreement + // renders as "null" instead of panicking; the three are built from the same spec, + // so they always line up in practice. + let value = key.data().fields().get(index).and_then(Option::as_ref); + let human = match fields.get(index) { + Some(nested) => human_string(&field.transform, &nested.field_type, value), + None => NULL.to_string(), + }; + form_urlencoded::Serializer::new(String::new()) + .append_pair(&field.name, &human) + .finish() + }) + .join("/") + } +} + +impl LocationGenerator for CometLocationGenerator { + fn generate_location(&self, partition_key: Option<&PartitionKey>, file_name: &str) -> String { + // `is_effectively_none` is iceberg-rust's own predicate: no key, or a key whose spec is + // unpartitioned (no fields, or every field a `void` transform). Matching it keeps the + // layout decision -- partition directory or not -- identical to + // `DefaultLocationGenerator`. + if PartitionKey::is_effectively_none(partition_key) { + format!("{}/{}", self.data_location, file_name) + } else { + format!( + "{}/{}/{}", + self.data_location, + self.partition_to_path(partition_key.unwrap()), + file_name + ) + } + } +} + +const NULL: &str = "null"; + +/// The value half of one `name=value` partition-path pair, as iceberg-java's +/// `Transform#toHumanString(Type, T)` renders it. +/// +/// Delegates to iceberg-rust's `Transform::to_human_string` and overrides only the arms where the +/// two disagree: +/// +/// | Iceberg type | iceberg-java | iceberg-rust | +/// |--------------------|-------------------------------|----------------------------------| +/// | `timestamp` | `1969-12-31T23:59:58.5` | `1969-12-31 23:59:58.500` | +/// | `timestamptz` | `1969-12-31T23:59:58.5+00:00` | panics for a negative value with a sub-second part; otherwise `1969-12-31 23:59:58.500 UTC` | +/// | `binary` / `fixed` | base64 | uppercase hex | +/// +/// The nanosecond timestamp types get the same treatment for the same reason. They are V3-only, so +/// `CometIcebergNativeWrite`'s format-version gate keeps them out of a native write today; the arms +/// exist so a future V3 write does not reintroduce the panic. +/// +/// Known remaining divergence, deliberately left delegating: `float` and `double`. Java renders +/// them with `Float.toString`/`Double.toString` (always a fractional digit, `E` notation outside +/// `[1e-3, 1e7)`), Rust with its own shortest representation, so `1.0` becomes `1` and `1.0E20` +/// becomes `100000000000000000000`. Porting Java's algorithm is a much larger piece of work than a +/// partition directory name warrants -- Comet's `cast(float as string)` needs the same port -- and +/// unlike `timestamptz` it does not panic. Iceberg deprecated float/double partitioning in 1.3. +fn human_string(transform: &Transform, field_type: &Type, value: Option<&Literal>) -> String { + // Java returns "null" for a null partition value regardless of transform or type, which also + // covers every `void` field: `void` produces no value, so this is the only arm it reaches. + let Some(primitive) = value.and_then(Literal::as_primitive_literal) else { + return NULL.to_string(); + }; + + // `year`/`month`/`day`/`hour` render the ordinal itself and never see a timestamp or binary + // field type (their result types are `int` and `date`), so they cannot collide with the arms + // below. iceberg-rust already mirrors `TransformUtil` for them. + match (field_type.as_primitive_type(), &primitive) { + (Some(PrimitiveType::Timestamp), PrimitiveLiteral::Long(micros)) => { + iso_timestamp(*micros, 6, false) + } + (Some(PrimitiveType::Timestamptz), PrimitiveLiteral::Long(micros)) => { + iso_timestamp(*micros, 6, true) + } + (Some(PrimitiveType::TimestampNs), PrimitiveLiteral::Long(nanos)) => { + iso_timestamp(*nanos, 9, false) + } + (Some(PrimitiveType::TimestamptzNs), PrimitiveLiteral::Long(nanos)) => { + iso_timestamp(*nanos, 9, true) + } + ( + Some(PrimitiveType::Binary | PrimitiveType::Fixed(_)), + PrimitiveLiteral::Binary(bytes), + ) => BASE64.encode(bytes), + _ => transform.to_human_string(field_type, value), + } +} + +/// Renders a sub-second count since the Unix epoch the way iceberg-java's +/// `DateTimeUtil.microsToIsoTimestamp[tz]` / `nanosToIsoTimestamp[tz]` do: +/// `DateTimeFormatter.ISO_LOCAL_DATE_TIME` over the UTC `LocalDateTime`, optionally followed by the +/// fixed `+00:00` offset the timestamptz formatter appends (the offset is always UTC, so +/// `appendOffset("+HH:MM:ss", "+00:00")` always emits its no-offset text verbatim). +/// +/// `subsecond_digits` is 6 for the microsecond types and 9 for the nanosecond ones. +/// +/// `ISO_LOCAL_DATE_TIME` always prints the seconds field (it is available on a `LocalDateTime`, so +/// its optional section is never dropped) and prints a fraction only when the nanosecond field is +/// non-zero, with trailing zeros stripped -- `appendFraction(NANO_OF_SECOND, 0, 9, true)` output- +/// scales a `stripTrailingZeros`ed `BigDecimal`. Half a second is therefore `.5`, not `.500000`. +fn iso_timestamp(value: i64, subsecond_digits: usize, with_zone: bool) -> String { + let per_second = 10i64.pow(subsecond_digits as u32); + let seconds = value.div_euclid(per_second); + let subsecond = value.rem_euclid(per_second); + let (year, month, day) = civil_from_days(seconds.div_euclid(SECONDS_PER_DAY)); + let second_of_day = seconds.rem_euclid(SECONDS_PER_DAY); + + let mut out = iso_year(year); + out.push_str(&format!( + "-{month:02}-{day:02}T{:02}:{:02}:{:02}", + second_of_day / 3_600, + (second_of_day / 60) % 60, + second_of_day % 60, + )); + if subsecond != 0 { + out.push('.'); + let digits = format!("{subsecond:0subsecond_digits$}"); + out.push_str(digits.trim_end_matches('0')); + } + if with_zone { + out.push_str("+00:00"); + } + out +} + +/// The year as `ISO_LOCAL_DATE` writes it: `appendValue(YEAR, 4, 10, SignStyle.EXCEEDS_PAD)`, so +/// four zero-padded digits inside `0..=9999`, a `+` sign above that, and `-` plus four zero-padded +/// digits below zero. +fn iso_year(year: i64) -> String { + if (0..=9999).contains(&year) { + format!("{year:04}") + } else if year > 9999 { + format!("+{year}") + } else { + format!("-{:04}", year.unsigned_abs()) + } +} + +/// Splits a day count since 1970-01-01 into `(year, month, day)` in the proleptic Gregorian +/// calendar (Howard Hinnant's `civil_from_days`). +/// +/// Hand-rolled rather than delegated to `chrono` because this must be total: every `i64` micros +/// value reaches a partition path, and `chrono`'s calendar stops at year 262143 while `i64` micros +/// reach year 292277. iceberg-java's `ChronoUnit.MICROS.addTo(EPOCH, micros)` has no such ceiling, +/// so a value past `chrono`'s range must still produce the same string, not an error we cannot +/// return from `generate_location` anyway. +fn civil_from_days(days: i64) -> (i64, u32, u32) { + // Shift the epoch to 0000-03-01 so leap days land at the end of the 400-year era. + let shifted = days + 719_468; + let era = if shifted >= 0 { + shifted + } else { + shifted - 146_096 + } / 146_097; + let day_of_era = (shifted - era * 146_097) as u64; // [0, 146096] + let year_of_era = + (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; // [0, 399] + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); // [0, 365] + let march_month = (5 * day_of_year + 2) / 153; // [0, 11], 0 = March + let day = (day_of_year - (153 * march_month + 2) / 5 + 1) as u32; // [1, 31] + let month = if march_month < 10 { + march_month + 3 + } else { + march_month - 9 + } as u32; // [1, 12] + let year = year_of_era as i64 + era * 400; + (if month <= 2 { year + 1 } else { year }, month, day) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use iceberg::spec::{ + NestedField, PartitionSpec, Schema as IcebergSchema, Struct as IcebergStruct, + }; + + use super::*; + + const MICROS_PER_DAY: i64 = SECONDS_PER_DAY * 1_000_000; + + fn timestamptz(micros: i64) -> String { + human_string( + &Transform::Identity, + &Type::Primitive(PrimitiveType::Timestamptz), + Some(&Literal::Primitive(PrimitiveLiteral::Long(micros))), + ) + } + + fn timestamp(micros: i64) -> String { + human_string( + &Transform::Identity, + &Type::Primitive(PrimitiveType::Timestamp), + Some(&Literal::Primitive(PrimitiveLiteral::Long(micros))), + ) + } + + // The whole point of the timestamptz override: `microseconds_to_datetimetz` takes + // `micros % 1_000_000` -- negative for a pre-epoch value -- casts it to `u32` and multiplies + // by 1000, then unwraps the `None` that `DateTime::from_timestamp` returns for the resulting + // out-of-range nanosecond count. Still present in iceberg-rust as of the pinned rev + // (665c64e); `nanoseconds_to_datetimetz` has the same shape. + #[test] + fn renders_pre_epoch_timestamptz_that_upstream_panics_on() { + assert_eq!(timestamptz(-1), "1969-12-31T23:59:59.999999+00:00"); + assert_eq!(timestamptz(-1_500_000), "1969-12-31T23:59:58.5+00:00"); + assert_eq!( + timestamptz(-MICROS_PER_DAY - 1), + "1969-12-30T23:59:59.999999+00:00" + ); + } + + // Values from Iceberg's own `TestTransformUtil`/`DateTimeUtil` behaviour: seconds are always + // printed, the fraction only when non-zero and with trailing zeros stripped. + #[test] + fn matches_java_iso_timestamp_formatting() { + assert_eq!(timestamptz(0), "1970-01-01T00:00:00+00:00"); + assert_eq!( + timestamptz(1_510_871_468_000_000), + "2017-11-16T22:31:08+00:00" + ); + assert_eq!(timestamptz(500_000), "1970-01-01T00:00:00.5+00:00"); + assert_eq!(timestamptz(100_000), "1970-01-01T00:00:00.1+00:00"); + assert_eq!(timestamptz(120_000), "1970-01-01T00:00:00.12+00:00"); + assert_eq!(timestamptz(123_456), "1970-01-01T00:00:00.123456+00:00"); + assert_eq!(timestamptz(1), "1970-01-01T00:00:00.000001+00:00"); + assert_eq!(timestamptz(10), "1970-01-01T00:00:00.00001+00:00"); + // Same rendering without the offset for the untagged type. + assert_eq!(timestamp(0), "1970-01-01T00:00:00"); + assert_eq!(timestamp(-1_500_000), "1969-12-31T23:59:58.5"); + assert_eq!(timestamp(123_456), "1970-01-01T00:00:00.123456"); + } + + #[test] + fn renders_nanosecond_timestamps() { + let ntz = human_string( + &Transform::Identity, + &Type::Primitive(PrimitiveType::TimestampNs), + Some(&Literal::Primitive(PrimitiveLiteral::Long(-1))), + ); + assert_eq!(ntz, "1969-12-31T23:59:59.999999999"); + let tz = human_string( + &Transform::Identity, + &Type::Primitive(PrimitiveType::TimestamptzNs), + Some(&Literal::Primitive(PrimitiveLiteral::Long(1_500_000_000))), + ); + assert_eq!(tz, "1970-01-01T00:00:01.5+00:00"); + } + + // `appendValue(YEAR, 4, 10, EXCEEDS_PAD)`: zero-padded to four digits, `+` above 9999, `-` + // plus four zero-padded digits below zero. Not `%04d`, which would render year -1 as "-001". + #[test] + fn renders_years_outside_the_four_digit_range() { + assert_eq!(iso_year(0), "0000"); + assert_eq!(iso_year(1), "0001"); + assert_eq!(iso_year(9999), "9999"); + assert_eq!(iso_year(10_000), "+10000"); + assert_eq!(iso_year(-1), "-0001"); + assert_eq!(iso_year(-10_000), "-10000"); + } + + // `i64` micros reach year 292277, past `chrono`'s year-262143 ceiling; iceberg-java has no + // such ceiling, so these must still render rather than fail. + #[test] + fn renders_timestamps_beyond_chronos_calendar() { + assert_eq!(timestamptz(i64::MAX), "+294247-01-10T04:00:54.775807+00:00"); + assert_eq!(timestamptz(i64::MIN), "-290308-12-21T19:59:05.224192+00:00"); + } + + #[test] + fn civil_from_days_matches_the_gregorian_calendar() { + assert_eq!(civil_from_days(0), (1970, 1, 1)); + assert_eq!(civil_from_days(-1), (1969, 12, 31)); + assert_eq!(civil_from_days(59), (1970, 3, 1)); + // 1972 was a leap year; 1900 was not. + assert_eq!(civil_from_days(789), (1972, 2, 29)); + assert_eq!(civil_from_days(-25_567), (1900, 1, 1)); + assert_eq!(civil_from_days(-25_509), (1900, 2, 28)); + assert_eq!(civil_from_days(-25_508), (1900, 3, 1)); + assert_eq!(civil_from_days(-719_468), (0, 3, 1)); + assert_eq!(civil_from_days(-719_528), (0, 1, 1)); + assert_eq!(civil_from_days(-719_529), (-1, 12, 31)); + } + + // Java base64-encodes binary and fixed partition values; iceberg-rust hex-encodes them. + #[test] + fn base64_encodes_binary_partition_values() { + for field_type in [PrimitiveType::Binary, PrimitiveType::Fixed(3)] { + let encoded = human_string( + &Transform::Identity, + &Type::Primitive(field_type), + Some(&Literal::Primitive(PrimitiveLiteral::Binary(vec![ + 0x00, 0x01, 0xff, + ]))), + ); + assert_eq!(encoded, "AAH/"); + } + } + + // Everything not overridden stays on iceberg-rust's renderer, which already mirrors + // `TransformUtil` for these. + #[test] + fn delegates_the_types_iceberg_rust_already_matches() { + let cases: Vec<(PrimitiveType, PrimitiveLiteral, &str)> = vec![ + ( + PrimitiveType::Boolean, + PrimitiveLiteral::Boolean(true), + "true", + ), + (PrimitiveType::Int, PrimitiveLiteral::Int(-7), "-7"), + (PrimitiveType::Long, PrimitiveLiteral::Long(-7), "-7"), + (PrimitiveType::Date, PrimitiveLiteral::Int(-1), "1969-12-31"), + ( + PrimitiveType::String, + PrimitiveLiteral::String("a b".to_string()), + "a b", + ), + ( + PrimitiveType::Decimal { + precision: 9, + scale: 2, + }, + PrimitiveLiteral::Int128(-105), + "-1.05", + ), + ]; + for (field_type, literal, expected) in cases { + let rendered = human_string( + &Transform::Identity, + &Type::Primitive(field_type.clone()), + Some(&Literal::Primitive(literal)), + ); + assert_eq!(rendered, expected, "type={field_type:?}"); + } + } + + #[test] + fn renders_a_missing_value_as_null() { + assert_eq!( + human_string( + &Transform::Identity, + &Type::Primitive(PrimitiveType::Timestamptz), + None + ), + "null" + ); + // A `void` field never carries a value, so it takes the same arm. + assert_eq!( + human_string( + &Transform::Void, + &Type::Primitive(PrimitiveType::Long), + None + ), + "null" + ); + } + + fn schema() -> Arc { + Arc::new( + IcebergSchema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "ts", Type::Primitive(PrimitiveType::Timestamptz)) + .into(), + NestedField::optional(3, "name", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .unwrap(), + ) + } + + fn generator(spec: &PartitionSpec, schema: &Arc) -> CometLocationGenerator { + CometLocationGenerator::try_new("file:/tmp/t/data".to_string(), spec, schema).unwrap() + } + + #[test] + fn unpartitioned_location_has_no_partition_directory() { + let schema = schema(); + let spec = PartitionSpec::builder(Arc::clone(&schema)).build().unwrap(); + let generator = generator(&spec, &schema); + assert_eq!( + generator.generate_location(None, "f.parquet"), + "file:/tmp/t/data/f.parquet" + ); + // A key whose spec is unpartitioned is treated as no key at all, matching + // `DefaultLocationGenerator`. + let key = PartitionKey::new(spec, Arc::clone(&schema), IcebergStruct::empty()); + assert_eq!( + generator.generate_location(Some(&key), "f.parquet"), + "file:/tmp/t/data/f.parquet" + ); + } + + #[test] + fn partitioned_location_escapes_names_and_values() { + let schema = schema(); + let spec = PartitionSpec::builder(Arc::clone(&schema)) + .with_spec_id(1) + .add_partition_field("ts", "ts_id", Transform::Identity) + .unwrap() + .add_partition_field("name", "the name", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let generator = generator(&spec, &schema); + let key = PartitionKey::new( + spec, + Arc::clone(&schema), + IcebergStruct::from_iter([ + Some(Literal::Primitive(PrimitiveLiteral::Long(-1_500_000))), + Some(Literal::Primitive(PrimitiveLiteral::String( + "a/b c".to_string(), + ))), + ]), + ); + assert_eq!( + generator.generate_location(Some(&key), "f.parquet"), + "file:/tmp/t/data/ts_id=1969-12-31T23%3A59%3A58.5%2B00%3A00/the+name=a%2Fb+c/f.parquet" + ); + } + + // A `void` field alongside a real one keeps its slot in the path with the literal value + // "null", as `Transform#toHumanString` renders it on the Java side. + #[test] + fn void_field_renders_as_null_in_the_path() { + let schema = schema(); + let spec = PartitionSpec::builder(Arc::clone(&schema)) + .with_spec_id(2) + .add_partition_field("ts", "ts_id", Transform::Void) + .unwrap() + .add_partition_field("id", "id_id", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let generator = generator(&spec, &schema); + let key = PartitionKey::new( + spec, + Arc::clone(&schema), + IcebergStruct::from_iter([None, Some(Literal::Primitive(PrimitiveLiteral::Int(5)))]), + ); + assert_eq!( + generator.generate_location(Some(&key), "f.parquet"), + "file:/tmp/t/data/ts_id=null/id_id=5/f.parquet" + ); + } +} diff --git a/native/core/src/execution/operators/iceberg_write.rs b/native/core/src/execution/operators/iceberg_write.rs index f7275ff1957..1fefcf7dd44 100644 --- a/native/core/src/execution/operators/iceberg_write.rs +++ b/native/core/src/execution/operators/iceberg_write.rs @@ -52,9 +52,7 @@ use iceberg::spec::{ Struct as IcebergStruct, StructType, }; use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; -use iceberg::writer::file_writer::location_generator::{ - DefaultFileNameGenerator, DefaultLocationGenerator, -}; +use iceberg::writer::file_writer::location_generator::DefaultFileNameGenerator; use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; use iceberg::writer::file_writer::ParquetWriterBuilder; use iceberg::writer::partitioning::clustered_writer::ClusteredWriter; @@ -72,10 +70,11 @@ use datafusion_comet_proto::spark_operator::{ use crate::cloud::s3::credential_bridge::AccessMode; use crate::execution::operators::iceberg_common::load_file_io; +use crate::execution::operators::iceberg_partition_path::CometLocationGenerator; /// Builder chain instantiated once per task and handed to the partitioning wrapper. type IcebergDataFileWriterBuilder = - DataFileWriterBuilder; + DataFileWriterBuilder; /// Native Iceberg write operator. Owns the parsed Iceberg schema/spec and the parquet writer /// properties; at task execution it builds the iceberg-rust writer stack, drains the upstream @@ -310,8 +309,15 @@ async fn run_write_task( AccessMode::Write, )?; - let location_generator = - DefaultLocationGenerator::with_data_location(common.data_location.clone()); + // Resolves the write's partition type once per task, before any data is written, so a spec the + // location generator could not render fails the task cleanly rather than panicking inside the + // infallible `LocationGenerator::generate_location`. + let location_generator = CometLocationGenerator::try_new( + common.data_location.clone(), + &partition_spec, + &iceberg_schema, + ) + .map_err(iceberg_err)?; let file_name_generator = DefaultFileNameGenerator::new( file_name_prefix(partition_id, task_attempt_id, &common.operation_id), None, @@ -617,9 +623,13 @@ async fn encode_data_files_as_manifest( operation_id, ); let output_file = memory_io.new_output(&path).map_err(iceberg_err)?; - let mut manifest_writer = - ManifestWriterBuilder::new(output_file, None, iceberg_schema, (*partition_spec).clone()) - .build_v2_data(); + let mut manifest_writer = ManifestWriterBuilder::new( + output_file, + None, + iceberg_schema, + manifest_partition_spec(&partition_spec), + ) + .build_v2_data(); for data_file in data_files { manifest_writer .add_file(data_file, 0) @@ -638,6 +648,37 @@ async fn encode_data_files_as_manifest( Ok(bytes.to_vec()) } +/// The partition spec the per-task transport manifest is encoded against. +/// +/// Normally the write's own spec, but a spec that `PartitionSpec::is_unpartitioned` accepts can +/// still carry fields: iceberg-java's `UpdatePartitionSpec` keeps a dropped partition field in a +/// format-version-1 spec as a `void` transform to preserve its field id, and "unpartitioned" means +/// *every* field is `void` on both the Java and Rust sides, not that there are none. `run_write_task` +/// routes such a write through `UnpartitionedWriter`, which stamps every data file with an empty +/// partition struct, while `ManifestWriter` derives its partition summaries from the spec's fields +/// and `zip_eq`s the two -- panicking across the JNI boundary on the length mismatch +/// (apache/datafusion-comet#5691). Encoding against a field-less spec of the same id makes the two +/// agree. +/// +/// Nothing downstream loses information. The JVM re-reads this manifest with the spec embedded in +/// its own Avro metadata, then rebuilds each `DataFile` against the table's real output spec, whose +/// `DataFiles.Builder` drops partition data outright for an unpartitioned spec -- so the manifest +/// that reaches storage carries exactly what iceberg-java's own writer would have committed for a +/// `void`-only spec (a null per `void` field, filled in by `PartitionData.get` returning null past +/// the end of its backing array). +/// +/// Dropping the fields also skips the `partition_type` resolution `ManifestWriter` would otherwise +/// do, which fails once the `void` field's source column has itself been dropped from the schema +/// (apache/datafusion-comet#5693). +fn manifest_partition_spec(partition_spec: &PartitionSpecRef) -> PartitionSpec { + if partition_spec.is_unpartitioned() { + // A no-op when the spec already has no fields, which is the common case. + PartitionSpec::unpartition_spec().with_spec_id(partition_spec.spec_id()) + } else { + (**partition_spec).clone() + } +} + fn build_output_batch(manifest_bytes: Vec, output_schema: &SchemaRef) -> DFResult { let array: ArrayRef = Arc::new(BinaryArray::from(vec![manifest_bytes.as_slice()])); RecordBatch::try_new(Arc::clone(output_schema), vec![array]).map_err(DataFusionError::from) @@ -1180,6 +1221,221 @@ mod tests { let err = decorate_batch_with_field_ids(batch, &target).unwrap_err(); assert!(format!("{err}").contains("column count mismatch")); } + + /// A spec whose only partition field is a `void` transform, as iceberg-java's + /// `UpdatePartitionSpec` leaves a format-version-1 spec after `DROP PARTITION FIELD`. + /// Built by deserialising the spec JSON rather than through `PartitionSpec::builder`, + /// which is how the real path builds it (`parse_partition_spec`) and which is also the + /// only way to reach a `void` field whose source column is gone. + fn void_spec(source_id: i32) -> PartitionSpec { + serde_json::from_str(&format!( + r#"{{"spec-id":2,"fields":[ + {{"source-id":{source_id},"field-id":1000, + "name":"region_part","transform":"void"}}]}}"# + )) + .unwrap() + } + + /// Regression for apache/datafusion-comet#5691. `is_unpartitioned()` accepts a spec whose + /// fields are all `void`, so the write routes through `UnpartitionedWriter` and every data + /// file gets an empty partition struct -- while `ManifestWriter` derives one partition + /// summary per spec field and `zip_eq`s the two, which used to panic (`itertools: + /// .zip_eq() reached end of one iterator before the other`) instead of returning an error. + #[tokio::test] + async fn void_only_spec_write_round_trips_through_the_manifest() { + let temp_dir = TempDir::new().unwrap(); + let data_location = format!("file://{}", temp_dir.path().display()); + let schema = iceberg_user_schema(); + // source-id 2 is `region`, still present in the schema. + let spec = void_spec(2); + assert!(spec.is_unpartitioned() && !spec.fields().is_empty()); + let common = common( + data_location.clone(), + serde_json::to_string(&spec).unwrap(), + serde_json::to_string(&schema).unwrap(), + ProtoIcebergWriterMode::IcebergWriterUnpartitioned, + ); + + let schema_arc = Arc::new(schema); + let spec_arc = Arc::new(spec); + let data_files = run_write_task( + input_stream(vec![batch(&[1, 2], &["us", "eu"])]), + Arc::clone(&common), + Arc::clone(&schema_arc), + Arc::clone(&spec_arc), + ProtoIcebergWriterMode::IcebergWriterUnpartitioned, + WriterProperties::builder().build(), + Some(0), + Some(0), + Time::default(), + ) + .await + .unwrap(); + assert_eq!(data_files.len(), 1); + // No partition directory, matching iceberg-java's `UnpartitionedDataWriter`: the file + // sits directly under the data location. + let file_path = data_files[0].file_path().to_string(); + let relative = file_path + .strip_prefix(&format!("{data_location}/")) + .unwrap_or_else(|| panic!("{file_path} is not under {data_location}")); + assert!( + !relative.contains('/'), + "unexpected directory in {relative}" + ); + + let manifest_bytes = encode_data_files_as_manifest( + data_files, + schema_arc, + Arc::clone(&spec_arc), + Some(0), + Some(0), + &common.operation_id, + ) + .await + .unwrap(); + let manifest = Manifest::parse_avro(&manifest_bytes).unwrap(); + assert_eq!(manifest.entries().len(), 1); + assert_eq!(manifest.entries()[0].data_file().record_count(), 2); + // The transport manifest is encoded against a field-less spec of the same id, so the + // JVM reads back an empty partition struct -- which is what `DataFiles.Builder` would + // have kept for this spec anyway. + assert_eq!(manifest.metadata().partition_spec().spec_id(), 2); + assert!(manifest.metadata().partition_spec().fields().is_empty()); + } + + /// Regression for apache/datafusion-comet#5693, the same shape one step further along: + /// the `void` field's source column has since been dropped from the schema, which used to + /// fail the write with "No column with source column id 9 in schema" from the manifest + /// encode. Nothing needs that column -- a `void` field contributes no partition value and + /// no partition directory. + #[tokio::test] + async fn void_field_with_a_dropped_source_column_still_writes() { + let temp_dir = TempDir::new().unwrap(); + let data_location = format!("file://{}", temp_dir.path().display()); + let schema = iceberg_user_schema(); + let spec = void_spec(9); + let common = common( + data_location, + serde_json::to_string(&spec).unwrap(), + serde_json::to_string(&schema).unwrap(), + ProtoIcebergWriterMode::IcebergWriterUnpartitioned, + ); + + let schema_arc = Arc::new(schema); + let spec_arc = Arc::new(spec); + let data_files = run_write_task( + input_stream(vec![batch(&[1], &["us"])]), + Arc::clone(&common), + Arc::clone(&schema_arc), + Arc::clone(&spec_arc), + ProtoIcebergWriterMode::IcebergWriterUnpartitioned, + WriterProperties::builder().build(), + Some(0), + Some(0), + Time::default(), + ) + .await + .unwrap(); + let manifest_bytes = encode_data_files_as_manifest( + data_files, + schema_arc, + spec_arc, + Some(0), + Some(0), + &common.operation_id, + ) + .await + .unwrap(); + assert_eq!( + Manifest::parse_avro(&manifest_bytes) + .unwrap() + .entries() + .len(), + 1 + ); + } + + /// Regression for apache/datafusion-comet#5694. A pre-epoch `timestamptz` partition value + /// used to panic while iceberg-rust rendered the partition directory name + /// (`microseconds_to_datetimetz` unwraps a `None` for a negative sub-second remainder); + /// Comet now renders the path itself, in iceberg-java's format. + #[tokio::test] + async fn pre_epoch_timestamptz_partition_gets_a_java_shaped_directory() { + use arrow::array::TimestampMicrosecondArray; + + let temp_dir = TempDir::new().unwrap(); + let data_location = format!("file://{}", temp_dir.path().display()); + let schema = Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "ts", Type::Primitive(PrimitiveType::Timestamptz)) + .into(), + ]) + .build() + .unwrap(); + let spec = PartitionSpec::builder(Arc::new(schema.clone())) + .with_spec_id(1) + .add_partition_field("ts", "ts_part", Transform::Identity) + .unwrap() + .build() + .unwrap(); + let common = common( + data_location, + serde_json::to_string(&spec).unwrap(), + serde_json::to_string(&schema).unwrap(), + ProtoIcebergWriterMode::IcebergWriterFanout, + ); + + // 1969-12-31T23:59:58.5Z: negative micros with a sub-second remainder. + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new( + "ts", + DataType::Timestamp( + arrow::datatypes::TimeUnit::Microsecond, + Some("UTC".into()), + ), + false, + ), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&arrow_schema), + vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new( + TimestampMicrosecondArray::from(vec![-1_500_000i64]).with_timezone("UTC"), + ), + ], + ) + .unwrap(); + + let data_files = run_write_task( + Box::pin(RecordBatchStreamAdapter::new( + arrow_schema, + futures::stream::iter(vec![Ok::<_, DataFusionError>(batch)]), + )), + common, + Arc::new(schema), + Arc::new(spec), + ProtoIcebergWriterMode::IcebergWriterFanout, + WriterProperties::builder().build(), + Some(0), + Some(0), + Time::default(), + ) + .await + .unwrap(); + + assert_eq!(data_files.len(), 1); + assert!( + data_files[0] + .file_path() + .contains("/ts_part=1969-12-31T23%3A59%3A58.5%2B00%3A00/"), + "unexpected partition directory in {}", + data_files[0].file_path() + ); + } } } diff --git a/native/core/src/execution/operators/mod.rs b/native/core/src/execution/operators/mod.rs index a669678d0ef..fbbdd0b0f2d 100644 --- a/native/core/src/execution/operators/mod.rs +++ b/native/core/src/execution/operators/mod.rs @@ -31,6 +31,7 @@ pub use expand::ExpandExec; mod explode; pub use explode::ExplodeExec; mod iceberg_common; +mod iceberg_partition_path; mod iceberg_scan; mod iceberg_write; pub use iceberg_write::IcebergWriteExec; diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index 8095e6feb05..ecc494ba15b 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -1033,6 +1033,124 @@ class CometIcebergWriteActionSuite } } + // iceberg-java renders a `timestamptz` partition value with + // `DateTimeUtil.microsToIsoTimestamptz` and a `binary` one with base64; iceberg-rust's + // `partition_to_path` renders the first as a `chrono` `DateTime` (space separator, " UTC" + // suffix) -- and outright panics on a pre-1970 value with a sub-second part, since + // `microseconds_to_datetimetz` casts a negative remainder to `u32` + // (apache/datafusion-comet#5694) -- and the second as uppercase hex. Comet renders the path + // itself; this pins the result against the layout iceberg-java's own writer produces. + test("native acceleration: timestamptz and binary partition paths match iceberg-java") { + assumeNativeAcceleration() + withIcebergCatalog { warehouseDir => + // A UTC session zone makes the stored micros of each literal exact, so the expected + // directory names below are not a function of the machine's zone. + withSQLConf("spark.sql.session.timeZone" -> "UTC") { + Seq("ts_path_native", "ts_path_jvm").foreach { table => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table (id INT, ts TIMESTAMP, bin BINARY) + USING iceberg PARTITIONED BY (ts, bin) + """) + } + // Pre-epoch with a sub-second part (the panic case), pre-epoch on a whole second, the + // epoch itself, and a post-epoch microsecond value. + val values = + "(1, TIMESTAMP '1969-12-31 23:59:58.5', X'0001FF'), " + + "(2, TIMESTAMP '1969-12-31 23:59:58', X'00'), " + + "(3, TIMESTAMP '1970-01-01 00:00:00', X''), " + + "(4, TIMESTAMP '2024-04-01 19:25:00.123456', X'FF')" + + assertNativeWriteEngages("ts_path_native", Seq(1, 2, 3, 4)) { + spark.sql(s"INSERT INTO $catalog.$ns.ts_path_native VALUES $values") + } + spark.sql(s"INSERT INTO $catalog.$ns.ts_path_jvm VALUES $values") + + val nativeDirs = partitionDirs(warehouseDir, "ts_path_native") + assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"), s"native: $nativeDirs") + assert( + nativeDirs.contains("ts=1969-12-31T23%3A59%3A58.5%2B00%3A00/bin=AAH%2F"), + s"native: $nativeDirs") + assert( + nativeDirs.contains("ts=1970-01-01T00%3A00%3A00%2B00%3A00/bin="), + s"native: $nativeDirs") + assert( + nativeDirs.contains("ts=2024-04-01T19%3A25%3A00.123456%2B00%3A00/bin=%2Fw%3D%3D"), + s"native: $nativeDirs") + + // Values round-trip through both readers regardless of how the path was spelled. + Seq("true", "false").foreach { cometEnabled => + withSQLConf(CometConf.COMET_ENABLED.key -> cometEnabled) { + val rows = + spark.sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_native ORDER BY id").collect() + assert( + rows.toSeq == + spark + .sql(s"SELECT id, ts FROM $catalog.$ns.ts_path_jvm ORDER BY id") + .collect() + .toSeq, + s"comet=$cometEnabled: $rows") + } + } + } + } + } + + // iceberg-java's `UpdatePartitionSpec` keeps a dropped partition field in a format-version-1 + // spec as a `void` transform so its field id survives, and `PartitionSpec#isUnpartitioned` is + // "every field is void", not "no fields". The next write therefore runs through the + // unpartitioned writer -- which stamps an empty partition struct -- against a spec whose fields + // are non-empty, and iceberg-rust's `ManifestWriter` used to `zip_eq` the two and panic across + // the JNI boundary (apache/datafusion-comet#5691). Dropping the source column afterwards then + // broke the manifest's `partition_type` resolution as well (apache/datafusion-comet#5693). + test("native acceleration: writes after a V1 partition field is dropped match iceberg-java") { + assumeNativeAcceleration() + withIcebergCatalog { _ => + Seq("evolved_native", "evolved_jvm").foreach { table => + spark.sql(s""" + CREATE TABLE $catalog.$ns.$table (id INT, region STRING) + USING iceberg TBLPROPERTIES ('format-version'='1') + """) + } + + def evolve(table: String, write: (String, Seq[Int]) => Unit): Unit = { + write("(1, 'us')", Seq(1)) + spark.sql(s"ALTER TABLE $catalog.$ns.$table ADD PARTITION FIELD region AS region_part") + write("(2, 'eu')", Seq(1, 2)) + spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP PARTITION FIELD region_part") + // The spec is now void-only: the #5691 shape. + write("(3, 'ap')", Seq(1, 2, 3)) + spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP COLUMN region") + // The void field's source column is gone: the #5693 shape. + write("(4)", Seq(1, 2, 3, 4)) + } + + evolve( + "evolved_native", + (row, expectedIds) => + assertNativeWriteEngages("evolved_native", expectedIds) { + spark.sql(s"INSERT INTO $catalog.$ns.evolved_native VALUES $row") + }) + evolve( + "evolved_jvm", + (row, _) => spark.sql(s"INSERT INTO $catalog.$ns.evolved_jvm VALUES $row")) + + def rows(table: String): Seq[Row] = + spark.sql(s"SELECT * FROM $catalog.$ns.$table ORDER BY id").collect().toSeq + assert(rows("evolved_native") == Seq(Row(1), Row(2), Row(3), Row(4))) + assert(rows("evolved_native") == rows("evolved_jvm")) + + // The committed manifests carry the same partition summaries the JVM writer produced. + def partitionSummaries(table: String): Seq[Row] = spark + .sql(s"SELECT partition_spec_id, partition_summaries FROM $catalog.$ns.$table.manifests" + + " ORDER BY partition_spec_id") + .collect() + .toSeq + assert( + partitionSummaries("evolved_native") == partitionSummaries("evolved_jvm"), + s"native: ${partitionSummaries("evolved_native")}") + } + } + test("native acceleration: fanout writer handles unsorted partitioned input") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => @@ -1312,20 +1430,10 @@ class CometIcebergWriteActionSuite // Every partition directory the native writer produced exists in the JVM writer's layout // and vice versa, so the two tables are byte-for-byte compatible in their data locations. - def partitionDirs(table: String): Set[String] = { - val location = warehouseDir.toURI.toString.stripSuffix("/") - spark - .sql(s"SELECT file_path FROM $catalog.$ns.$table.files") - .collect() - .map(_.getString(0)) - .map { path => - val relative = path.stripPrefix(location).split("/data/", 2)(1) - relative.substring(0, relative.lastIndexOf('/')) - } - .toSet - } - val nativeDirs = partitionDirs("escaped_native") - assert(nativeDirs == partitionDirs("escaped_jvm"), s"native layout: $nativeDirs") + val nativeDirs = partitionDirs(warehouseDir, "escaped_native") + assert( + nativeDirs == partitionDirs(warehouseDir, "escaped_jvm"), + s"native layout: $nativeDirs") assert(nativeDirs.contains("region=a%2Fb") && nativeDirs.contains("region=c%23d")) assert(nativeDirs.contains("region=g+h") && nativeDirs.contains("region=*-._")) @@ -1797,6 +1905,24 @@ class CometIcebergWriteActionSuite snapshot.plans.mkString("\n--\n")) } + /** + * The partition directory of every committed data file, relative to the table's `data/` + * 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`. + */ + private def partitionDirs(warehouseDir: File, tableName: String): Set[String] = { + val location = warehouseDir.toURI.toString.stripSuffix("/") + spark + .sql(s"SELECT file_path FROM $catalog.$ns.$tableName.files") + .collect() + .map(_.getString(0)) + .map { path => + val relative = path.stripPrefix(location).split("/data/", 2)(1) + relative.substring(0, relative.lastIndexOf('/')) + } + .toSet + } + private def assertRows(tableName: String, expectedIds: Seq[Int]): Unit = { val ids = spark .sql(s"SELECT id FROM $catalog.$ns.$tableName ORDER BY id") From 09e8a320cf9640ad9243ed03ae01180afe834f81 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 6 Sep 2026 09:25:06 -0600 Subject: [PATCH 2/2] fix: qualify partition-path parity by Iceberg version in the new tests Both new regression tests asserted iceberg-java parity unconditionally, which fails on the older Iceberg runtimes the Spark 3.4/3.5/4.0 profiles pin. The timestamp directory comparison against the JVM writer only holds on Iceberg 1.8+. Iceberg 1.5.2 (Spark 3.4) rendered a `timestamptz` partition value with `ChronoUnit.MICROS.addTo(EPOCH, micros).toString()`, spelling the same instant `1969-12-31T23:59:58.500Z` rather than `1969-12-31T23:59:58.5+00:00`, and left the field name unescaped; 1.8 moved to `DateTimeUtil.microsToIsoTimestamptz` and started escaping the name. Comet targets the 1.8+ spelling on every profile, so gate that one assertion and qualify the claim in both the module docs and iceberg-writes.md. The pinned expectations for what Comet writes, and the readback, stay unconditional. Writing after the `void` field's source column is dropped needs Iceberg 1.11+ on either path: `PartitionSpec.partitionType` NPEs on the missing source before 1.10, and `PartitionSpec.javaClasses` still does on 1.10 because `getResultType(null)` returns null. 1.11 substitutes `UnknownType` in both. That failure is in the driver-side commit, which the native path shares with the stock one, so normalising the transport manifest cannot rescue the older runtimes. Split it into its own version-gated test so the void-only coverage that every runtime can commit stays unconditional. Verified locally: CometIcebergWriteActionSuite passes on the spark-3.4 (Iceberg 1.5.2), spark-3.5 (1.8.1), spark-4.0 (1.10.0) and spark-4.1 (1.11.0) profiles, as do the other Iceberg suites in the `scans` group on 3.4 and 4.0. --- .../user-guide/latest/iceberg-writes.md | 11 ++- .../operators/iceberg_partition_path.rs | 9 +++ .../comet/CometIcebergWriteActionSuite.scala | 79 ++++++++++++++----- 3 files changed, 76 insertions(+), 23 deletions(-) diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md index d4159dae6a6..62a4df1f128 100644 --- a/docs/source/user-guide/latest/iceberg-writes.md +++ b/docs/source/user-guide/latest/iceberg-writes.md @@ -251,12 +251,15 @@ a data file but not what any reader computes from it: differences (iceberg-java checks the target file size every 1000 rows and names files `---`; iceberg-rust checks per batch and uses a process-local counter). -- Partition directory names match iceberg-java's `PartitionSpec.partitionToPath` for every +- Partition directory names match iceberg-java 1.8+'s `PartitionSpec.partitionToPath` for every partition type except `float` and `double`, where the value is rendered with Rust's shortest representation instead of `Float.toString` / `Double.toString` (`f=1` where iceberg-java writes - `f=1.0`). Distinct partition values still get distinct directories, and no reader parses these - names — files are resolved through committed manifests. Iceberg deprecated float and double - partitioning in 1.3. + `f=1.0`). 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 1.8+ spelling on every profile. Distinct partition values still get + distinct directories in all cases, and no reader parses these names — files are resolved through + committed manifests. Iceberg deprecated float and double partitioning in 1.3. - Compressed page bytes are implementation-defined: the codec and any explicit level are translated, but parquet-rs and parquet-mr embed different encoder implementations and defaults (zstd default levels, LZ4 framing), so byte-identical output is not achievable even diff --git a/native/core/src/execution/operators/iceberg_partition_path.rs b/native/core/src/execution/operators/iceberg_partition_path.rs index ea609a59b4e..a2cc5a95a32 100644 --- a/native/core/src/execution/operators/iceberg_partition_path.rs +++ b/native/core/src/execution/operators/iceberg_partition_path.rs @@ -124,6 +124,15 @@ const NULL: &str = "null"; /// The value half of one `name=value` partition-path pair, as iceberg-java's /// `Transform#toHumanString(Type, T)` renders it. /// +/// "as iceberg-java renders it" means iceberg-java 1.8 or later, which is what Comet targets on +/// every profile. Iceberg 1.5.x -- still the pinned runtime for the Spark 3.4 profile -- rendered +/// `timestamp` and `timestamptz` with `LocalDateTime.toString()` and `OffsetDateTime.toString()` +/// instead of the `DateTimeUtil.microsToIsoTimestamp[tz]` formatters 1.8 switched to, so it spells +/// the same instant `1969-12-31T23:59:58.500Z` rather than `1969-12-31T23:59:58.5+00:00`. (1.5.x +/// also left the field *name* unescaped, which no Comet version has reproduced either.) Every other +/// type renders identically from 1.5 through 1.11, and the partition values themselves are +/// unaffected -- only the directory name, which nothing parses. +/// /// Delegates to iceberg-rust's `Transform::to_human_string` and overrides only the arms where the /// two disagree: /// diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala index ecc494ba15b..9b0d8f412cd 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala @@ -1040,6 +1040,14 @@ class CometIcebergWriteActionSuite // `microseconds_to_datetimetz` casts a negative remainder to `u32` // (apache/datafusion-comet#5694) -- and the second as uppercase hex. Comet renders the path // itself; this pins the result against the layout iceberg-java's own writer produces. + // + // The byte-for-byte comparison against the JVM writer holds on Iceberg 1.8+ only. Iceberg 1.5.x + // (the Spark 3.4 profile) rendered a `timestamptz` partition value with + // `ChronoUnit.MICROS.addTo(EPOCH, micros).toString()`, i.e. `OffsetDateTime.toString()`, which + // spells the same instant `1969-12-31T23:59:58.500Z` rather than `1969-12-31T23:59:58.5+00:00`; + // 1.8 switched it to `microsToIsoTimestamptz` (and started escaping the field name too). Comet + // targets the 1.8+ spelling on every profile. The values Comet writes are the same either way, + // so the pinned expectations and the readback below stay unconditional. test("native acceleration: timestamptz and binary partition paths match iceberg-java") { assumeNativeAcceleration() withIcebergCatalog { warehouseDir => @@ -1066,7 +1074,9 @@ class CometIcebergWriteActionSuite spark.sql(s"INSERT INTO $catalog.$ns.ts_path_jvm VALUES $values") val nativeDirs = partitionDirs(warehouseDir, "ts_path_native") - assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"), s"native: $nativeDirs") + if (icebergVersionAtLeast(1, 8)) { + assert(nativeDirs == partitionDirs(warehouseDir, "ts_path_jvm"), s"native: $nativeDirs") + } assert( nativeDirs.contains("ts=1969-12-31T23%3A59%3A58.5%2B00%3A00/bin=AAH%2F"), s"native: $nativeDirs") @@ -1100,12 +1110,40 @@ class CometIcebergWriteActionSuite // "every field is void", not "no fields". The next write therefore runs through the // unpartitioned writer -- which stamps an empty partition struct -- against a spec whose fields // are non-empty, and iceberg-rust's `ManifestWriter` used to `zip_eq` the two and panic across - // the JNI boundary (apache/datafusion-comet#5691). Dropping the source column afterwards then - // broke the manifest's `partition_type` resolution as well (apache/datafusion-comet#5693). + // the JNI boundary (apache/datafusion-comet#5691). test("native acceleration: writes after a V1 partition field is dropped match iceberg-java") { assumeNativeAcceleration() + assertDroppedPartitionFieldParity("evolved", dropSourceColumn = false) + } + + // Dropping the `void` field's source column afterwards broke the transport manifest's + // `partition_type` resolution too (apache/datafusion-comet#5693). Only Iceberg 1.11+ can commit + // this at all, on either path: `PartitionSpec.partitionType` NPEs on the missing source before + // 1.10, and `PartitionSpec.javaClasses` still does on 1.10 (`getResultType(null)` returns null). + // 1.11 substitutes `UnknownType` in both. The driver-side commit is shared with the stock path, + // so normalising the native manifest cannot rescue the older runtimes. + test( + "native acceleration: writes after a V1 partition source column is dropped match " + + "iceberg-java") { + assumeNativeAcceleration() + assume( + icebergVersionAtLeast(1, 11), + "Iceberg < 1.11 cannot commit a dropped partition source") + assertDroppedPartitionFieldParity("evolved_dropped", dropSourceColumn = true) + } + + /** + * Walks the format-version-1 partition-spec evolution from #5691 twice -- once through the + * native writer, once through iceberg-java's -- and compares the results. `dropSourceColumn` + * extends the walk with the #5693 stage, which drops the `void` field's source column before + * writing again. + */ + private def assertDroppedPartitionFieldParity( + prefix: String, + dropSourceColumn: Boolean): Unit = { + val (nativeTable, jvmTable) = (s"${prefix}_native", s"${prefix}_jvm") withIcebergCatalog { _ => - Seq("evolved_native", "evolved_jvm").foreach { table => + Seq(nativeTable, jvmTable).foreach { table => spark.sql(s""" CREATE TABLE $catalog.$ns.$table (id INT, region STRING) USING iceberg TBLPROPERTIES ('format-version'='1') @@ -1119,35 +1157,38 @@ class CometIcebergWriteActionSuite spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP PARTITION FIELD region_part") // The spec is now void-only: the #5691 shape. write("(3, 'ap')", Seq(1, 2, 3)) - spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP COLUMN region") - // The void field's source column is gone: the #5693 shape. - write("(4)", Seq(1, 2, 3, 4)) + if (dropSourceColumn) { + spark.sql(s"ALTER TABLE $catalog.$ns.$table DROP COLUMN region") + write("(4)", Seq(1, 2, 3, 4)) + } } evolve( - "evolved_native", + nativeTable, (row, expectedIds) => - assertNativeWriteEngages("evolved_native", expectedIds) { - spark.sql(s"INSERT INTO $catalog.$ns.evolved_native VALUES $row") + assertNativeWriteEngages(nativeTable, expectedIds) { + spark.sql(s"INSERT INTO $catalog.$ns.$nativeTable VALUES $row") }) - evolve( - "evolved_jvm", - (row, _) => spark.sql(s"INSERT INTO $catalog.$ns.evolved_jvm VALUES $row")) + evolve(jvmTable, (row, _) => spark.sql(s"INSERT INTO $catalog.$ns.$jvmTable VALUES $row")) def rows(table: String): Seq[Row] = spark.sql(s"SELECT * FROM $catalog.$ns.$table ORDER BY id").collect().toSeq - assert(rows("evolved_native") == Seq(Row(1), Row(2), Row(3), Row(4))) - assert(rows("evolved_native") == rows("evolved_jvm")) + val expected = + if (dropSourceColumn) Seq(Row(1), Row(2), Row(3), Row(4)) + else Seq(Row(1, "us"), Row(2, "eu"), Row(3, "ap")) + assert(rows(nativeTable) == expected, s"native: ${rows(nativeTable)}") + assert(rows(nativeTable) == rows(jvmTable)) // The committed manifests carry the same partition summaries the JVM writer produced. def partitionSummaries(table: String): Seq[Row] = spark - .sql(s"SELECT partition_spec_id, partition_summaries FROM $catalog.$ns.$table.manifests" + - " ORDER BY partition_spec_id") + .sql( + s"SELECT partition_spec_id, partition_summaries FROM $catalog.$ns.$table.manifests" + + " ORDER BY partition_spec_id") .collect() .toSeq assert( - partitionSummaries("evolved_native") == partitionSummaries("evolved_jvm"), - s"native: ${partitionSummaries("evolved_native")}") + partitionSummaries(nativeTable) == partitionSummaries(jvmTable), + s"native: ${partitionSummaries(nativeTable)}") } }