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 200ee6593 fix(arrow): map Iceberg Binary to Arrow Binary instead of
LargeBinary (#2710)
200ee6593 is described below
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