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-2710-0ebe240a0f73a3efca2ad4a98326b06ccfd2e30e in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 200ee6593dee84654a3b98aed724c7d094aa18cf Author: Liang-Chi Hsieh <[email protected]> AuthorDate: Tue Nov 3 22:57:48 2026 +0000 fix(arrow): map Iceberg Binary to Arrow Binary instead of LargeBinary (#2710) * fix(arrow): map Iceberg Binary to Arrow Binary instead of LargeBinary `type_to_arrow_type` mapped `PrimitiveType::Binary` to `DataType::LargeBinary`, which is inconsistent with both the Java implementation (`ArrowSchemaUtil` maps Iceberg Binary to Arrow `Binary`) and parquet-rs, which reads an unannotated Parquet `BYTE_ARRAY` column as `DataType::Binary` by default. The mismatch forces an unnecessary cast when reading Binary columns and diverges from the rest of the ecosystem. Map `PrimitiveType::Binary` to `DataType::Binary`. The reverse conversion (`arrow_schema_to_schema`), the arrow-to-literal path, and the bucket transform already accept both `Binary` and `LargeBinary`, so externally produced `LargeBinary` columns keep working. Closes #2698 * fix(writer): accept LargeBinary equality columns after the Binary mapping change With Iceberg Binary now mapped to Arrow Binary, the equality delete projector builds its output batch against a Binary schema while reusing the input arrays as-is, so batches carrying LargeBinaryArray equality columns (accepted before) failed RecordBatch::try_new. Cast LargeBinary to Binary in RecordBatchProjector::project_batch and add a regression test. Also drop redundant `.as_slice()` in test fixtures. Co-authored-by: Isaac <[email protected]> * fix(arrow): accept LargeBinary source columns in partition value calculation PartitionValueCalculator builds its result struct against the partition Arrow type, which now uses Binary for Iceberg binary fields, while the identity transform passes the projected LargeBinary column through, so StructArray::try_new failed. Move the LargeBinary -> Binary cast from RecordBatchProjector::project_batch into project_column so both callers get columns matching the projected schema. This also lets truncate partitions on binary fields accept LargeBinary input. Co-authored-by: Isaac <[email protected]> --------- Co-authored-by: Isaac <[email protected]> --- .../src/arrow/partition_value_calculator.rs | 64 +++++++++++++++++- crates/iceberg/src/arrow/record_batch_projector.rs | 18 ++++- crates/iceberg/src/arrow/schema.rs | 9 +-- .../writer/base_writer/equality_delete_writer.rs | 76 +++++++++++++++++++++- .../src/writer/file_writer/parquet_writer.rs | 2 +- 5 files changed, 160 insertions(+), 9 deletions(-) diff --git a/crates/iceberg/src/arrow/partition_value_calculator.rs b/crates/iceberg/src/arrow/partition_value_calculator.rs index c8c917779..142de42cd 100644 --- a/crates/iceberg/src/arrow/partition_value_calculator.rs +++ b/crates/iceberg/src/arrow/partition_value_calculator.rs @@ -163,7 +163,7 @@ impl PartitionValueCalculator { mod tests { use std::sync::Arc; - use arrow_array::{Int32Array, RecordBatch, StringArray}; + use arrow_array::{BinaryArray, Int32Array, LargeBinaryArray, RecordBatch, StringArray}; use arrow_schema::{Field, Schema as ArrowSchema}; use super::*; @@ -220,6 +220,68 @@ mod tests { assert_eq!(id_partition.value(2), 30); } + #[test] + fn test_partition_calculator_binary_partition_accepts_large_binary_input() { + let table_schema = Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::optional(2, "key", Type::Primitive(PrimitiveType::Binary)).into(), + ]) + .build() + .unwrap(); + + let partition_spec = PartitionSpecBuilder::new(Arc::new(table_schema.clone())) + .add_partition_field("key", "key_identity", Transform::Identity) + .unwrap() + .add_partition_field("key", "key_trunc", Transform::Truncate(2)) + .unwrap() + .build() + .unwrap(); + let calculator = PartitionValueCalculator::try_new(&partition_spec, &table_schema).unwrap(); + + let arrow_schema = Arc::new(ArrowSchema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("key", DataType::LargeBinary, true), + ])); + let batch = RecordBatch::try_new(arrow_schema, vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(LargeBinaryArray::from_opt_vec(vec![ + Some(b"abc"), + None, + Some(b"z"), + ])), + ]) + .unwrap(); + + let result = calculator.calculate(&batch).unwrap(); + + let DataType::Struct(expected_fields) = calculator.partition_arrow_type() else { + unreachable!() + }; + let expected = StructArray::try_new( + expected_fields.clone(), + vec![ + Arc::new(BinaryArray::from_opt_vec(vec![ + Some(b"abc"), + None, + Some(b"z"), + ])), + Arc::new(BinaryArray::from_opt_vec(vec![ + Some(b"ab"), + None, + Some(b"z"), + ])), + ], + None, + ) + .unwrap(); + assert_eq!( + result.as_any().downcast_ref::<StructArray>().unwrap(), + &expected + ); + } + #[test] fn test_partition_calculator_unpartitioned_error() { let table_schema = Schema::builder() diff --git a/crates/iceberg/src/arrow/record_batch_projector.rs b/crates/iceberg/src/arrow/record_batch_projector.rs index 823ef3188..ca559933f 100644 --- a/crates/iceberg/src/arrow/record_batch_projector.rs +++ b/crates/iceberg/src/arrow/record_batch_projector.rs @@ -19,6 +19,7 @@ use std::sync::Arc; use arrow_array::{Array, ArrayRef, RecordBatch, StructArray, make_array}; use arrow_buffer::NullBuffer; +use arrow_cast::cast; use arrow_schema::{DataType, Field, FieldRef, Fields, Schema, SchemaRef}; use crate::arrow::schema::{schema_to_arrow_schema, try_get_field_id_from_metadata}; @@ -160,8 +161,21 @@ impl RecordBatchProjector { /// Do projection with columns pub fn project_column(&self, batch: &[ArrayRef]) -> Result<Vec<ArrayRef>> { let mut columns = Vec::with_capacity(self.field_indices.len()); - for index_vec in &self.field_indices { - columns.push(Self::get_column_by_field_index(batch, index_vec)?); + for (index_vec, field) in self + .field_indices + .iter() + .zip(self.projected_schema.fields()) + { + let column = Self::get_column_by_field_index(batch, index_vec)?; + let column = match (column.data_type(), field.data_type()) { + (DataType::LargeBinary, DataType::Binary) => cast(&column, &DataType::Binary) + .map_err(|err| { + invalid_data!("Failed to cast column {} to Binary", field.name()) + .with_source(err) + })?, + _ => column, + }; + columns.push(column); } Ok(columns) } diff --git a/crates/iceberg/src/arrow/schema.rs b/crates/iceberg/src/arrow/schema.rs index 91b8d687a..24fc93f03 100644 --- a/crates/iceberg/src/arrow/schema.rs +++ b/crates/iceberg/src/arrow/schema.rs @@ -890,7 +890,8 @@ impl SchemaVisitor for ToArrowSchemaConverter { .map(DataType::FixedSizeBinary) .unwrap_or(DataType::LargeBinary), )), - PrimitiveType::Binary | PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { + PrimitiveType::Binary => Ok(ArrowSchemaOrFieldOrType::Type(DataType::Binary)), + PrimitiveType::Geometry(_) | PrimitiveType::Geography(_) => { Ok(ArrowSchemaOrFieldOrType::Type(DataType::LargeBinary)) } } @@ -2015,8 +2016,8 @@ mod tests { false, "12", ), - simple_field("l", DataType::LargeBinary, false, "13"), - simple_field("o", DataType::LargeBinary, false, "22"), + simple_field("l", DataType::Binary, false, "13"), + simple_field("o", DataType::Binary, false, "22"), simple_field("m", DataType::FixedSizeBinary(10), false, "11"), simple_field( "list", @@ -2044,7 +2045,7 @@ mod tests { "fixed_list", DataType::List(Arc::new(simple_field( "element", - DataType::LargeBinary, + DataType::Binary, false, "26", ))), diff --git a/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs b/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs index 84c10b176..f62016add 100644 --- a/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs +++ b/crates/iceberg/src/writer/base_writer/equality_delete_writer.rs @@ -763,7 +763,7 @@ mod test { ) .unwrap(), ) as ArrayRef; - let col13 = Arc::new(arrow_array::LargeBinaryArray::from_opt_vec(vec![ + let col13 = Arc::new(arrow_array::BinaryArray::from_opt_vec(vec![ Some(b"one"), Some(b""), Some(b"zzzz"), @@ -785,6 +785,80 @@ mod test { Ok(()) } + #[tokio::test] + async fn test_equality_delete_accepts_large_binary_input() -> Result<(), anyhow::Error> { + let temp_dir = TempDir::new().unwrap(); + 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( + Schema::builder() + .with_schema_id(1) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::optional(2, "key", Type::Primitive(PrimitiveType::Binary)).into(), + ]) + .build() + .unwrap(), + ); + let config = EqualityDeleteWriterConfig::new(vec![2], schema.clone()).unwrap(); + let delete_arrow_schema = config.projected_arrow_schema_ref().clone(); + let delete_schema = arrow_schema_to_schema(&delete_arrow_schema).unwrap(); + + let pb = + ParquetWriterBuilder::new(WriterProperties::builder().build(), Arc::new(delete_schema)); + let rolling_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + pb, + file_io.clone(), + location_gen, + file_name_gen, + ); + let mut equality_delete_writer = + EqualityDeleteFileWriterBuilder::new(rolling_writer_builder, config) + .build(None) + .await?; + + let input_schema = Arc::new(arrow_schema::Schema::new(vec![ + Field::new("id", DataType::Int32, false).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "1".to_string(), + )])), + Field::new("key", DataType::LargeBinary, true).with_metadata(HashMap::from([( + PARQUET_FIELD_ID_META_KEY.to_string(), + "2".to_string(), + )])), + ])); + let to_write = RecordBatch::try_new(input_schema, vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef, + Arc::new(arrow_array::LargeBinaryArray::from_opt_vec(vec![ + Some(b"one"), + None, + Some(b"zzzz"), + ])) as ArrayRef, + ]) + .unwrap(); + equality_delete_writer.write(to_write).await?; + let res = equality_delete_writer.close().await?; + assert_eq!(res.len(), 1); + + let expected = RecordBatch::try_new(delete_arrow_schema, vec![Arc::new( + arrow_array::BinaryArray::from_opt_vec(vec![Some(b"one"), None, Some(b"zzzz")]), + ) as ArrayRef]) + .unwrap(); + check_parquet_data_file_with_equality_delete_write( + &file_io, + &res.into_iter().next().unwrap(), + &expected, + ) + .await; + + Ok(()) + } + #[tokio::test] async fn test_equality_delete_with_nullable_field() -> Result<(), anyhow::Error> { // prepare data diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index 1470b310b..485a94200 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -1377,7 +1377,7 @@ mod tests { None, Some("d"), ])) as ArrayRef; - let col6 = Arc::new(arrow_array::LargeBinaryArray::from_opt_vec(vec