This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3327-3c952e1719a4902334e44b648698999cf3a4214d in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 1f3bc34058e3f3e45336bc5428f37b92649a608b Author: Feng Zhang <[email protected]> AuthorDate: Fri Oct 9 17:29:50 2026 +0000 fix(spec): treat -0.0 and 0.0 as distinct float literals (#3327) * fix(spec): treat -0.0 and 0.0 as distinct float literals PrimitiveLiteral equality and ordering now follow Java's Float.compare and Double.compare: -0.0 is less than 0.0 and all NaNs are equal. This stops the partitioned writers, the record batch splitter and the delete file index from merging -0.0 and 0.0 identity partitions. Closes #3325 * Address review: fail closed in PrimitiveLiteral fallback, test IN - Handle AboveMax and BelowMin explicitly and return None when the fallback arm sees two values of the same variant, with a debug assertion, instead of calling them equal. - Cite the spec's partition equality rule and document the variant_index ordering invariant. - Stop pinning the cross-variant order in tests. - Test IN with signed zeros in the expression evaluator. --- crates/iceberg/public-api.txt | 5 +- .../src/arrow/record_batch_partition_splitter.rs | 75 ++++++++++++++++++- crates/iceberg/src/delete_file_index.rs | 32 ++++++++ .../src/expr/visitors/expression_evaluator.rs | 70 +++++++++++++++++ crates/iceberg/src/spec/values/primitive.rs | 80 +++++++++++++++++++- crates/iceberg/src/spec/values/tests.rs | 59 +++++++++++++++ .../src/writer/partitioning/clustered_writer.rs | 84 ++++++++++++++++++++- .../src/writer/partitioning/fanout_writer.rs | 87 +++++++++++++++++++++- 8 files changed, 483 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index bd5539534..8a34bbf7b 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1693,16 +1693,15 @@ impl core::clone::Clone for iceberg::spec::PrimitiveLiteral pub fn iceberg::spec::PrimitiveLiteral::clone(&self) -> iceberg::spec::PrimitiveLiteral impl core::cmp::Eq for iceberg::spec::PrimitiveLiteral impl core::cmp::PartialEq for iceberg::spec::PrimitiveLiteral -pub fn iceberg::spec::PrimitiveLiteral::eq(&self, other: &iceberg::spec::PrimitiveLiteral) -> bool +pub fn iceberg::spec::PrimitiveLiteral::eq(&self, other: &Self) -> bool impl core::cmp::PartialOrd for iceberg::spec::PrimitiveLiteral -pub fn iceberg::spec::PrimitiveLiteral::partial_cmp(&self, other: &iceberg::spec::PrimitiveLiteral) -> core::option::Option<core::cmp::Ordering> +pub fn iceberg::spec::PrimitiveLiteral::partial_cmp(&self, other: &Self) -> core::option::Option<core::cmp::Ordering> impl core::convert::From<iceberg::spec::Datum> for iceberg::spec::PrimitiveLiteral pub fn iceberg::spec::PrimitiveLiteral::from(value: iceberg::spec::Datum) -> Self impl core::fmt::Debug for iceberg::spec::PrimitiveLiteral pub fn iceberg::spec::PrimitiveLiteral::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::hash::Hash for iceberg::spec::PrimitiveLiteral pub fn iceberg::spec::PrimitiveLiteral::hash<__H: core::hash::Hasher>(&self, state: &mut __H) -impl core::marker::StructuralPartialEq for iceberg::spec::PrimitiveLiteral pub enum iceberg::spec::PrimitiveType pub iceberg::spec::PrimitiveType::Binary pub iceberg::spec::PrimitiveType::Boolean diff --git a/crates/iceberg/src/arrow/record_batch_partition_splitter.rs b/crates/iceberg/src/arrow/record_batch_partition_splitter.rs index 0b679040c..af3c99a97 100644 --- a/crates/iceberg/src/arrow/record_batch_partition_splitter.rs +++ b/crates/iceberg/src/arrow/record_batch_partition_splitter.rs @@ -209,7 +209,7 @@ impl RecordBatchPartitionSplitter { mod tests { use std::sync::Arc; - use arrow_array::{Int32Array, RecordBatch, StringArray}; + use arrow_array::{Float64Array, Int32Array, RecordBatch, StringArray}; use arrow_schema::DataType; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; @@ -474,4 +474,77 @@ mod tests { assert_eq!(ids, vec![3, 3]); assert_eq!(names, vec!["d", "f"]); } + + #[test] + fn test_record_batch_partition_split_signed_zero() { + let schema = Arc::new( + Schema::builder() + .with_fields(vec![ + NestedField::required( + 1, + "id", + Type::Primitive(crate::spec::PrimitiveType::Int), + ) + .into(), + NestedField::required( + 2, + "d", + Type::Primitive(crate::spec::PrimitiveType::Double), + ) + .into(), + ]) + .build() + .unwrap(), + ); + let partition_spec = Arc::new( + PartitionSpecBuilder::new(schema.clone()) + .with_spec_id(1) + .add_unbound_field( + UnboundPartitionField::builder() + .source_ids(vec![2]) + .name("d".to_string()) + .transform(Transform::Identity) + .build() + .unwrap(), + ) + .unwrap() + .build() + .unwrap(), + ); + let partition_splitter = RecordBatchPartitionSplitter::try_new_with_computed_values( + schema.clone(), + partition_spec, + ) + .expect("Failed to create splitter"); + + let arrow_schema = Arc::new(schema_to_arrow_schema(&schema).unwrap()); + let batch = RecordBatch::try_new(arrow_schema, vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(Float64Array::from(vec![-0.0, 0.0, -0.0])), + ]) + .expect("Failed to create RecordBatch"); + + let mut partitions: Vec<(Struct, Vec<i32>)> = partition_splitter + .split(&batch) + .expect("Failed to split RecordBatch") + .into_iter() + .map(|(key, batch)| { + let ids = batch + .column(0) + .as_any() + .downcast_ref::<Int32Array>() + .unwrap() + .values() + .to_vec(); + (key.data().clone(), ids) + }) + .collect(); + partitions.sort_by_key(|(_, ids)| ids[0]); + + // -0.0 and 0.0 are different partition values, as in iceberg-java. + assert_eq!(partitions, vec![ + (Struct::from_iter([Some(Literal::double(-0.0))]), vec![1, 3]), + (Struct::from_iter([Some(Literal::double(0.0))]), vec![2]), + ]); + } } diff --git a/crates/iceberg/src/delete_file_index.rs b/crates/iceberg/src/delete_file_index.rs index 8bae43941..01b4773f6 100644 --- a/crates/iceberg/src/delete_file_index.rs +++ b/crates/iceberg/src/delete_file_index.rs @@ -592,6 +592,38 @@ mod tests { assert!(actual_paths_to_apply_for_different_spec.is_empty()); } + #[test] + fn test_partition_scoped_deletes_distinguish_signed_zero() { + let neg_zero = Struct::from_iter([Some(Literal::double(-0.0))]); + let pos_zero = Struct::from_iter([Some(Literal::double(0.0))]); + let spec_id = 1; + + let delete_contexts: Vec<DeleteFileContext> = [ + build_partitioned_eq_delete(&neg_zero, spec_id), + build_partitioned_pos_delete(&neg_zero, spec_id), + ] + .iter() + .map(|delete| DeleteFileContext { + manifest_entry: build_added_manifest_entry(4, delete).into(), + partition_spec_id: spec_id, + }) + .collect(); + + let delete_file_index = PopulatedDeleteFileIndex::new(delete_contexts).unwrap(); + + let deletes_for_neg_zero = delete_file_index + .get_deletes_for_data_file(&build_partitioned_data_file(&neg_zero, spec_id), Some(0)) + .unwrap(); + assert_eq!(deletes_for_neg_zero.len(), 2); + + // -0.0 and 0.0 are different partitions, as in iceberg-java, so deletes written + // for one do not apply to data files in the other. + let deletes_for_pos_zero = delete_file_index + .get_deletes_for_data_file(&build_partitioned_data_file(&pos_zero, spec_id), Some(0)) + .unwrap(); + assert!(deletes_for_pos_zero.is_empty()); + } + #[test] fn test_pos_delete_with_referenced_data_file_applies_only_to_that_file() { let data_file_a = build_unpartitioned_data_file(); diff --git a/crates/iceberg/src/expr/visitors/expression_evaluator.rs b/crates/iceberg/src/expr/visitors/expression_evaluator.rs index 31ce94a99..d9c2e1b57 100644 --- a/crates/iceberg/src/expr/visitors/expression_evaluator.rs +++ b/crates/iceberg/src/expr/visitors/expression_evaluator.rs @@ -596,6 +596,76 @@ mod tests { Ok(()) } + #[test] + fn test_expr_eq_signed_zero() -> Result<()> { + let case_sensitive = true; + let (partition_spec, schema) = create_partition_spec(PrimitiveType::Float)?; + + let data_file = DataFile { + partition: Struct::from_iter([Some(Literal::float(-0.0_f32))]), + ..create_data_file_float() + }; + + // -0.0 and 0.0 are different values, as in iceberg-java's evaluator. + for (literal, expected) in [(0.0_f32, false), (-0.0_f32, true)] { + let predicate = Predicate::Binary(BinaryExpression::new( + PredicateOperator::Eq, + Reference::new("a"), + Datum::float(literal), + )) + .bind(schema.clone(), case_sensitive)?; + + let expression_evaluator = create_expression_evaluator( + partition_spec.clone(), + &schema, + &predicate, + case_sensitive, + )?; + + assert_eq!(expression_evaluator.eval(&data_file)?, expected); + } + + Ok(()) + } + + #[test] + fn test_expr_in_signed_zero() -> Result<()> { + let case_sensitive = true; + let (partition_spec, schema) = create_partition_spec(PrimitiveType::Float)?; + + let data_file = DataFile { + partition: Struct::from_iter([Some(Literal::float(-0.0_f32))]), + ..create_data_file_float() + }; + + // -0.0 and 0.0 are different values, so `IN` keeps both and matches only its own zero. + for (literals, expected) in [ + ([0.0_f32, 1.0_f32], false), + ([-0.0_f32, 1.0_f32], true), + ([0.0_f32, -0.0_f32], true), + ] { + let predicate = Predicate::Set(SetExpression::new( + PredicateOperator::In, + Reference::new("a"), + FnvHashSet::from_iter(literals.map(Datum::float)), + )) + .bind(schema.clone(), case_sensitive)?; + + assert!(matches!(&predicate, BoundPredicate::Set(expr) if expr.literals().len() == 2)); + + let expression_evaluator = create_expression_evaluator( + partition_spec.clone(), + &schema, + &predicate, + case_sensitive, + )?; + + assert_eq!(expression_evaluator.eval(&data_file)?, expected); + } + + Ok(()) + } + #[test] fn test_expr_greater_than_or_eq() -> Result<()> { let case_sensitive = true; diff --git a/crates/iceberg/src/spec/values/primitive.rs b/crates/iceberg/src/spec/values/primitive.rs index 43d5c48c5..b76978bd5 100644 --- a/crates/iceberg/src/spec/values/primitive.rs +++ b/crates/iceberg/src/spec/values/primitive.rs @@ -17,10 +17,21 @@ //! Primitive literal types -use ordered_float::OrderedFloat; +use std::cmp::Ordering; + +use ordered_float::{FloatCore, OrderedFloat}; /// Values present in iceberg type -#[derive(Clone, Debug, PartialOrd, PartialEq, Hash, Eq)] +/// +/// `Float` and `Double` compare as Java's `Float.compare` and `Double.compare` do: `-0.0` is +/// less than `0.0`, and every NaN is the same value, greater than all others. iceberg-java +/// compares partition values this way, so `-0.0` and `0.0` are different partitions. The +/// [spec](https://iceberg.apache.org/spec/#scan-planning) states the same rule: floating point +/// partition values are equal if their IEEE 754 bit layouts are equal, with NaNs normalized. +// The derived `Hash` gives `-0.0` and `0.0` the same hash. That is coarser than `eq`, which is +// allowed: equal values still hash alike. +#[allow(clippy::derived_hash_with_manual_eq)] +#[derive(Clone, Debug, Hash, Eq)] pub enum PrimitiveLiteral { /// 0x00 for false, non-zero byte for true Boolean(bool), @@ -46,7 +57,72 @@ pub enum PrimitiveLiteral { BelowMin, } +impl PartialEq for PrimitiveLiteral { + fn eq(&self, other: &Self) -> bool { + self.partial_cmp(other) == Some(Ordering::Equal) + } +} + +impl PartialOrd for PrimitiveLiteral { + fn partial_cmp(&self, other: &Self) -> Option<Ordering> { + match (self, other) { + (Self::Boolean(a), Self::Boolean(b)) => a.partial_cmp(b), + (Self::Int(a), Self::Int(b)) => a.partial_cmp(b), + (Self::Long(a), Self::Long(b)) => a.partial_cmp(b), + (Self::Float(a), Self::Float(b)) => Some(float_cmp(a, b)), + (Self::Double(a), Self::Double(b)) => Some(float_cmp(a, b)), + (Self::String(a), Self::String(b)) => a.partial_cmp(b), + (Self::Binary(a), Self::Binary(b)) => a.partial_cmp(b), + (Self::Int128(a), Self::Int128(b)) => a.partial_cmp(b), + (Self::UInt128(a), Self::UInt128(b)) => a.partial_cmp(b), + (Self::AboveMax, Self::AboveMax) | (Self::BelowMin, Self::BelowMin) => { + Some(Ordering::Equal) + } + // Different variants order by declaration, as the derived impl did. A variant that + // lacks an arm above lands here against itself; fail closed instead of calling the + // two values equal. + _ => { + debug_assert_ne!(std::mem::discriminant(self), std::mem::discriminant(other)); + match self.variant_index().cmp(&other.variant_index()) { + Ordering::Equal => None, + ordering => Some(ordering), + } + } + } + } +} + +/// Compares floats as Java's `Float.compare` and `Double.compare` do. `OrderedFloat` already +/// treats every NaN as one value above all others; it only lacks `-0.0` before `0.0`. +fn float_cmp<T: FloatCore>(a: &OrderedFloat<T>, b: &OrderedFloat<T>) -> Ordering { + a.cmp(b).then_with(|| { + if a.is_nan() { + Ordering::Equal + } else { + b.is_sign_negative().cmp(&a.is_sign_negative()) + } + }) +} + impl PrimitiveLiteral { + /// Must follow the declaration order of the variants: `partial_cmp` uses it to order + /// different variants the way the derived `PartialOrd` did. + fn variant_index(&self) -> u8 { + match self { + Self::Boolean(_) => 0, + Self::Int(_) => 1, + Self::Long(_) => 2, + Self::Float(_) => 3, + Self::Double(_) => 4, + Self::String(_) => 5, + Self::Binary(_) => 6, + Self::Int128(_) => 7, + Self::UInt128(_) => 8, + Self::AboveMax => 9, + Self::BelowMin => 10, + } + } + /// Returns true if the Literal represents a primitive type /// that can be a NaN, and that it's value is NaN pub fn is_nan(&self) -> bool { diff --git a/crates/iceberg/src/spec/values/tests.rs b/crates/iceberg/src/spec/values/tests.rs index c756bc3f8..61f83efee 100644 --- a/crates/iceberg/src/spec/values/tests.rs +++ b/crates/iceberg/src/spec/values/tests.rs @@ -17,6 +17,8 @@ //! Tests for Iceberg value types +use std::collections::HashSet; + use apache_avro::to_value; use apache_avro::types::Value; use ordered_float::OrderedFloat; @@ -1534,6 +1536,63 @@ fn test_negative_zero_less_than_positive_zero() { } } +#[test] +fn test_signed_zero_literals_are_distinct() { + for (neg_zero, pos_zero) in [ + ( + PrimitiveLiteral::Float(OrderedFloat(-0.0)), + PrimitiveLiteral::Float(OrderedFloat(0.0)), + ), + ( + PrimitiveLiteral::Double(OrderedFloat(-0.0)), + PrimitiveLiteral::Double(OrderedFloat(0.0)), + ), + ] { + assert_ne!(neg_zero, pos_zero); + assert_eq!( + neg_zero.partial_cmp(&pos_zero), + Some(std::cmp::Ordering::Less) + ); + assert_eq!(HashSet::from([neg_zero, pos_zero]).len(), 2); + } + + assert_ne!(Datum::float(-0.0_f32), Datum::float(0.0_f32)); + assert_ne!(Datum::double(-0.0), Datum::double(0.0)); + + // A partition tuple is a `Struct`, which the partitioned writers use as a map key. + let partitions = HashSet::from([ + Struct::from_iter([Some(Literal::double(-0.0))]), + Struct::from_iter([Some(Literal::double(0.0))]), + ]); + assert_eq!(partitions.len(), 2); +} + +#[test] +fn test_nan_literals_are_equal() { + for (nan, other_nan) in [ + ( + PrimitiveLiteral::Float(OrderedFloat(f32::NAN)), + PrimitiveLiteral::Float(OrderedFloat(-f32::NAN)), + ), + ( + PrimitiveLiteral::Double(OrderedFloat(f64::NAN)), + PrimitiveLiteral::Double(OrderedFloat(-f64::NAN)), + ), + ] { + assert_eq!(nan, other_nan); + assert_eq!(nan.partial_cmp(&other_nan), Some(std::cmp::Ordering::Equal)); + assert_eq!(HashSet::from([nan, other_nan]).len(), 1); + } +} + +#[test] +fn test_primitive_literal_eq_across_variants() { + assert_eq!(PrimitiveLiteral::AboveMax, PrimitiveLiteral::AboveMax); + assert_eq!(PrimitiveLiteral::BelowMin, PrimitiveLiteral::BelowMin); + assert_ne!(PrimitiveLiteral::AboveMax, PrimitiveLiteral::BelowMin); + assert_ne!(PrimitiveLiteral::Int(1), PrimitiveLiteral::Long(1)); +} + /// Test Date deserialization from JSON as number (days since epoch). /// /// This reproduces the scenario from Iceberg Java's TestAddFilesProcedure where: diff --git a/crates/iceberg/src/writer/partitioning/clustered_writer.rs b/crates/iceberg/src/writer/partitioning/clustered_writer.rs index 1fd9738c4..21b52bd81 100644 --- a/crates/iceberg/src/writer/partitioning/clustered_writer.rs +++ b/crates/iceberg/src/writer/partitioning/clustered_writer.rs @@ -145,13 +145,14 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; - use arrow_array::{Int32Array, RecordBatch, StringArray}; + use arrow_array::{Float64Array, Int32Array, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::file::properties::WriterProperties; use tempfile::TempDir; use super::*; + use crate::arrow::schema_to_arrow_schema; use crate::io::FileIO; use crate::spec::{DataFileFormat, NestedField, PrimitiveType, Type}; use crate::writer::base_writer::data_file_writer::DataFileWriterBuilder; @@ -508,4 +509,85 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn test_clustered_writer_signed_zero_partitions() -> Result<()> { + let temp_dir = TempDir::new()?; + let file_io = FileIO::new_with_fs(); + let location_gen = DefaultLocationGenerator::with_data_location( + temp_dir.path().to_str().unwrap().to_string(), + ); + let file_name_gen = + DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet); + + let schema = Arc::new( + crate::spec::Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "d", Type::Primitive(PrimitiveType::Double)).into(), + ]) + .build()?, + ); + let partition_spec = crate::spec::PartitionSpec::builder(schema.clone()) + .add_partition_field("d", "d", crate::spec::Transform::Identity)? + .build()?; + + let partition_value = |d: f64| Struct::from_iter([Some(crate::spec::Literal::double(d))]); + let partition_key = + |d: f64| PartitionKey::new(partition_spec.clone(), schema.clone(), partition_value(d)); + + let arrow_schema = Arc::new(schema_to_arrow_schema(&schema)?); + let batch = |id: i32, d: f64| { + RecordBatch::try_new(arrow_schema.clone(), vec![ + Arc::new(Int32Array::from(vec![id])), + Arc::new(Float64Array::from(vec![d])), + ]) + }; + + let parquet_writer_builder = + ParquetWriterBuilder::new(WriterProperties::builder().build(), schema.clone()); + let rolling_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + file_io.clone(), + location_gen, + file_name_gen, + ); + + // -0.0 and 0.0 are different partition values, as in iceberg-java, so 0.0 right + // after -0.0 starts a new data file. + let mut writer = + ClusteredWriter::new(DataFileWriterBuilder::new(rolling_writer_builder.clone())); + writer.write(partition_key(-0.0), batch(1, -0.0)?).await?; + writer.write(partition_key(0.0), batch(2, 0.0)?).await?; + let partitions_written: Vec<Struct> = writer + .close() + .await? + .into_iter() + .map(|data_file| data_file.partition) + .collect(); + assert_eq!(partitions_written, vec![ + partition_value(-0.0), + partition_value(0.0) + ]); + + // 0.0 is not the closed -0.0 partition, so this input is still sorted. + let mut writer = ClusteredWriter::new(DataFileWriterBuilder::new(rolling_writer_builder)); + writer.write(partition_key(-0.0), batch(1, -0.0)?).await?; + writer.write(partition_key(1.0), batch(3, 1.0)?).await?; + writer.write(partition_key(0.0), batch(2, 0.0)?).await?; + let partitions_written: Vec<Struct> = writer + .close() + .await? + .into_iter() + .map(|data_file| data_file.partition) + .collect(); + assert_eq!(partitions_written, vec![ + partition_value(-0.0), + partition_value(1.0), + partition_value(0.0) + ]); + + Ok(()) + } } diff --git a/crates/iceberg/src/writer/partitioning/fanout_writer.rs b/crates/iceberg/src/writer/partitioning/fanout_writer.rs index 5348fea15..2cdd86537 100644 --- a/crates/iceberg/src/writer/partitioning/fanout_writer.rs +++ b/crates/iceberg/src/writer/partitioning/fanout_writer.rs @@ -119,17 +119,18 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; - use arrow_array::{Int32Array, RecordBatch, StringArray}; + use arrow_array::{Float64Array, Int32Array, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; use parquet::file::properties::WriterProperties; use tempfile::TempDir; use super::*; + use crate::arrow::schema_to_arrow_schema; use crate::io::FileIO; use crate::spec::{ DataFileFormat, Literal, NestedField, PartitionKey, PartitionSpec, PrimitiveType, Struct, - Type, + Transform, Type, }; use crate::writer::base_writer::data_file_writer::DataFileWriterBuilder; use crate::writer::file_writer::ParquetWriterBuilder; @@ -381,4 +382,86 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn test_fanout_writer_signed_zero_partitions() -> Result<()> { + let temp_dir = TempDir::new()?; + let file_io = FileIO::new_with_fs(); + let location_gen = DefaultLocationGenerator::with_data_location( + temp_dir.path().to_str().unwrap().to_string(), + ); + let file_name_gen = + DefaultFileNameGenerator::new("test".to_string(), None, DataFileFormat::Parquet); + + let schema = Arc::new( + crate::spec::Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "d", Type::Primitive(PrimitiveType::Double)).into(), + ]) + .build()?, + ); + let partition_spec = PartitionSpec::builder(schema.clone()) + .add_partition_field("d", "d", Transform::Identity)? + .build()?; + + let partition_value_neg_zero = Struct::from_iter([Some(Literal::double(-0.0))]); + let partition_key_neg_zero = PartitionKey::new( + partition_spec.clone(), + schema.clone(), + partition_value_neg_zero.clone(), + ); + + let partition_value_pos_zero = Struct::from_iter([Some(Literal::double(0.0))]); + let partition_key_pos_zero = PartitionKey::new( + partition_spec.clone(), + schema.clone(), + partition_value_pos_zero.clone(), + ); + + let parquet_writer_builder = + ParquetWriterBuilder::new(WriterProperties::builder().build(), schema.clone()); + let rolling_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + file_io.clone(), + location_gen, + file_name_gen, + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_writer_builder); + + let mut writer = FanoutWriter::new(data_file_writer_builder); + + let arrow_schema = Arc::new(schema_to_arrow_schema(&schema)?); + let batch_neg_zero = RecordBatch::try_new(arrow_schema.clone(), vec![ + Arc::new(Int32Array::from(vec![1])), + Arc::new(Float64Array::from(vec![-0.0])), + ])?; + let batch_pos_zero = RecordBatch::try_new(arrow_schema.clone(), vec![ + Arc::new(Int32Array::from(vec![2])), + Arc::new(Float64Array::from(vec![0.0])), + ])?; + + writer.write(partition_key_neg_zero, batch_neg_zero).await?; + writer.write(partition_key_pos_zero, batch_pos_zero).await?; + + let data_files = writer.close().await?; + + // -0.0 and 0.0 are different partition values, as in iceberg-java, so each row + // goes to its own data file. + assert_eq!(data_files.len(), 2); + for data_file in &data_files { + assert_eq!(data_file.record_count, 1); + } + let partitions_found: std::collections::HashSet<Struct> = data_files + .iter() + .map(|data_file| data_file.partition.clone()) + .collect(); + assert_eq!( + partitions_found, + std::collections::HashSet::from([partition_value_neg_zero, partition_value_pos_zero]) + ); + + Ok(()) + } }
