This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1f3bc3405 fix(spec): treat -0.0 and 0.0 as distinct float literals
(#3327)
1f3bc3405 is described below
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(())
+ }
}
