comphead commented on code in PR #7198: URL: https://github.com/apache/datafusion-comet/pull/7198#discussion_r4417472377
########## native/core/src/execution/operators/iceberg_uuid.rs: ########## @@ -0,0 +1,819 @@ +// 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. + +//! Iceberg UUID strings use Java's `UUID.fromString` semantics, including its permissive +//! five-group fallback. Only Iceberg UUID fields use this conversion; `fixed(16)` stays binary. + +use std::sync::Arc; + +use arrow::array::{ + Array, ArrayRef, AsArray, FixedSizeBinaryBuilder, ListArray, MapArray, StructArray, +}; +use arrow::datatypes::{DataType, Field, SchemaRef}; +use arrow::record_batch::RecordBatch; +use datafusion::error::{DataFusionError, Result}; +use iceberg::spec::{PrimitiveType, Schema, Type}; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +/// Java parses UTF-16 code units, so supplementary-plane digits are deliberately excluded. +/// This table is pinned against Character.digit(char, 16) on JDK 17 and 21 over the entire BMP. +fn java_hex_digit(c: char) -> Option<u64> { + match c { + '0'..='9' => Some(c as u64 - '0' as u64), + 'a'..='f' => Some(c as u64 - 'a' as u64 + 10), + 'A'..='F' => Some(c as u64 - 'A' as u64 + 10), + '\u{ff21}'..='\u{ff26}' => Some(c as u64 - 0xff21 + 10), + '\u{ff41}'..='\u{ff46}' => Some(c as u64 - 0xff41 + 10), + _ => { + const ZEROS: &[u32] = &[ + 0x660, 0x6f0, 0x7c0, 0x966, 0x9e6, 0xa66, 0xae6, 0xb66, 0xbe6, 0xc66, 0xce6, 0xd66, + 0xde6, 0xe50, 0xed0, 0xf20, 0x1040, 0x1090, 0x17e0, 0x1810, 0x1946, 0x19d0, 0x1a80, + 0x1a90, 0x1b50, 0x1bb0, 0x1c40, 0x1c50, 0xa620, 0xa8d0, 0xa900, 0xa9d0, 0xa9f0, + 0xaa50, 0xabf0, 0xff10, + ]; + let value = c as u32; + ZEROS.iter().find_map(|&zero| { + (value >= zero && value < zero + 10).then(|| u64::from(value - zero)) + }) + } + } +} + +fn parse_java_uuid(value: &str) -> Option<[u8; 16]> { + // Avoid the group parser and Unicode handling for canonical ASCII strings. parse_str alone + // also accepts forms Java rejects, so require the canonical dash positions first. + let bytes = value.as_bytes(); + if bytes.len() == 36 && [8, 13, 18, 23].iter().all(|&i| bytes[i] == b'-') { + if let Ok(uuid) = uuid::Uuid::parse_str(value) { + return Some(*uuid.as_bytes()); + } + } + if value.encode_utf16().take(37).count() > 36 { + return None; + } + let mut groups = value.split('-'); + let mut parts = [0u64; 5]; + for part in &mut parts { + let group = groups.next()?; + let digits = group.strip_prefix('+').unwrap_or(group); + if digits.is_empty() { + return None; + } + let mut number = 0u64; + for c in digits.chars() { + number = number.checked_mul(16)?.checked_add(java_hex_digit(c)?)?; + // Long.parseLong rejects signed overflow *before* UUID masks the group width. + if number > i64::MAX as u64 { + return None; + } + } + *part = number; + } + if groups.next().is_some() { + return None; + } + let most = ((parts[0] & 0xffff_ffff) << 32) | ((parts[1] & 0xffff) << 16) | (parts[2] & 0xffff); + let least = ((parts[3] & 0xffff) << 48) | (parts[4] & 0xffff_ffff_ffff); + let mut out = [0u8; 16]; + out[..8].copy_from_slice(&most.to_be_bytes()); + out[8..].copy_from_slice(&least.to_be_bytes()); + Some(out) +} + +/// Constructed once per task. Subtrees without UUID retain the ordinary Arrow cast path. +pub(super) struct UuidAlignment(Vec<Conversion>); + +enum Conversion { + Cast, + Uuid, + Struct(Vec<Conversion>), + List(Box<Conversion>), + Map(Box<Conversion>), +} + +impl UuidAlignment { + pub(super) fn new(schema: &Schema, target: &SchemaRef) -> Self { + Self( + target + .fields() + .iter() + .map(|f| Conversion::new(schema, f)) + .collect(), + ) + } + + pub(super) fn convert(&self, batch: RecordBatch, target: &SchemaRef) -> Result<RecordBatch> { + if batch.num_columns() != self.0.len() { + return Err(DataFusionError::Execution( + "Iceberg write column count mismatch".into(), Review Comment: #7100 makes a top-level `unknown` column eligible by dropping `Null` columns in `decorate_batch_with_field_ids` before this count check, and this PR deletes that function. Whichever lands second needs `UuidAlignment::convert` to skip `Null` columns before comparing `batch.num_columns()` with `self.0.len()`, or a table with both a `uuid` and an `unknown` column fails with `column count mismatch`. `fall-back: uuid does not make an unknown column eligible` would then also start failing, since that table becomes eligible. Could the two of you coordinate, and turn that test into a native-write parity test for a `uuid` plus `unknown` table? This is from reading both diffs, not from a merged run. Separately, the old message carried "arrow batch has N columns but schema has M", which helped with the earlier pruning and `ReplaceData` projection bugs. Could the counts stay? ########## docs/source/user-guide/latest/iceberg.md: ########## @@ -494,7 +494,7 @@ A write is eligible only when ALL of the following hold: | Hadoop S3A settings for an `s3` / `s3a` data location | only `fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`, `endpoint.region`, and `path.style.access`, including their `fs.s3a.bucket.<data-bucket>.*` forms; any other effective `fs.s3a.*` setting falls back, apart from the ignored settings listed below | | Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location | the S3 endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`, `kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain settings parsed by the pinned iceberg-rust version, plus Comet's credential-provider class and built-in web-identity properties. When a custom provider is configured, its vendor-owned `s3.*` / `client.*` properties are also forwarded; unsupported Iceberg-defined S3 settings still fall back | | partition spec | any, except a `void` field whose source column was dropped beside a live field (see below) | Review Comment: The gate now declines a UUID partition or sort source, but this row still says "any, except a `void` field ...", and the rule appears only in the paragraph further down. Could the row, or a sort-order row, list it so the table shows every fallback? #7100 edits the same table. ########## native/core/src/execution/operators/iceberg_uuid.rs: ########## @@ -0,0 +1,819 @@ +// 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. + +//! Iceberg UUID strings use Java's `UUID.fromString` semantics, including its permissive +//! five-group fallback. Only Iceberg UUID fields use this conversion; `fixed(16)` stays binary. + +use std::sync::Arc; + +use arrow::array::{ + Array, ArrayRef, AsArray, FixedSizeBinaryBuilder, ListArray, MapArray, StructArray, +}; +use arrow::datatypes::{DataType, Field, SchemaRef}; +use arrow::record_batch::RecordBatch; +use datafusion::error::{DataFusionError, Result}; +use iceberg::spec::{PrimitiveType, Schema, Type}; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +/// Java parses UTF-16 code units, so supplementary-plane digits are deliberately excluded. +/// This table is pinned against Character.digit(char, 16) on JDK 17 and 21 over the entire BMP. +fn java_hex_digit(c: char) -> Option<u64> { + match c { + '0'..='9' => Some(c as u64 - '0' as u64), + 'a'..='f' => Some(c as u64 - 'a' as u64 + 10), + 'A'..='F' => Some(c as u64 - 'A' as u64 + 10), + '\u{ff21}'..='\u{ff26}' => Some(c as u64 - 0xff21 + 10), + '\u{ff41}'..='\u{ff46}' => Some(c as u64 - 0xff41 + 10), + _ => { + const ZEROS: &[u32] = &[ + 0x660, 0x6f0, 0x7c0, 0x966, 0x9e6, 0xa66, 0xae6, 0xb66, 0xbe6, 0xc66, 0xce6, 0xd66, + 0xde6, 0xe50, 0xed0, 0xf20, 0x1040, 0x1090, 0x17e0, 0x1810, 0x1946, 0x19d0, 0x1a80, + 0x1a90, 0x1b50, 0x1bb0, 0x1c40, 0x1c50, 0xa620, 0xa8d0, 0xa900, 0xa9d0, 0xa9f0, + 0xaa50, 0xabf0, 0xff10, + ]; + let value = c as u32; + ZEROS.iter().find_map(|&zero| { + (value >= zero && value < zero + 10).then(|| u64::from(value - zero)) + }) + } + } +} + +fn parse_java_uuid(value: &str) -> Option<[u8; 16]> { + // Avoid the group parser and Unicode handling for canonical ASCII strings. parse_str alone + // also accepts forms Java rejects, so require the canonical dash positions first. + let bytes = value.as_bytes(); + if bytes.len() == 36 && [8, 13, 18, 23].iter().all(|&i| bytes[i] == b'-') { + if let Ok(uuid) = uuid::Uuid::parse_str(value) { + return Some(*uuid.as_bytes()); + } + } + if value.encode_utf16().take(37).count() > 36 { + return None; + } + let mut groups = value.split('-'); + let mut parts = [0u64; 5]; + for part in &mut parts { + let group = groups.next()?; + let digits = group.strip_prefix('+').unwrap_or(group); + if digits.is_empty() { + return None; + } + let mut number = 0u64; + for c in digits.chars() { + number = number.checked_mul(16)?.checked_add(java_hex_digit(c)?)?; + // Long.parseLong rejects signed overflow *before* UUID masks the group width. + if number > i64::MAX as u64 { + return None; + } + } + *part = number; + } + if groups.next().is_some() { + return None; + } + let most = ((parts[0] & 0xffff_ffff) << 32) | ((parts[1] & 0xffff) << 16) | (parts[2] & 0xffff); + let least = ((parts[3] & 0xffff) << 48) | (parts[4] & 0xffff_ffff_ffff); + let mut out = [0u8; 16]; + out[..8].copy_from_slice(&most.to_be_bytes()); + out[8..].copy_from_slice(&least.to_be_bytes()); + Some(out) +} + +/// Constructed once per task. Subtrees without UUID retain the ordinary Arrow cast path. +pub(super) struct UuidAlignment(Vec<Conversion>); + +enum Conversion { + Cast, + Uuid, + Struct(Vec<Conversion>), + List(Box<Conversion>), + Map(Box<Conversion>), +} + +impl UuidAlignment { + pub(super) fn new(schema: &Schema, target: &SchemaRef) -> Self { + Self( + target + .fields() + .iter() + .map(|f| Conversion::new(schema, f)) + .collect(), + ) + } + + pub(super) fn convert(&self, batch: RecordBatch, target: &SchemaRef) -> Result<RecordBatch> { + if batch.num_columns() != self.0.len() { + return Err(DataFusionError::Execution( + "Iceberg write column count mismatch".into(), + )); + } + let columns = batch + .columns() + .iter() + .zip(target.fields()) + .zip(&self.0) + .map(|((array, field), conversion)| conversion.convert(array, field, None)) + .collect::<Result<Vec<_>>>()?; + Ok(RecordBatch::try_new(Arc::clone(target), columns)?) + } +} + +impl Conversion { + fn new(schema: &Schema, field: &Field) -> Self { + if field + .metadata() + .get(PARQUET_FIELD_ID_META_KEY) + .and_then(|id| id.parse().ok()) + .and_then(|id| schema.field_by_id(id)) + .is_some_and(|f| *f.field_type == Type::Primitive(PrimitiveType::Uuid)) + { + return Self::Uuid; + } + let result = match field.data_type() { + DataType::Struct(fields) => { + Self::Struct(fields.iter().map(|f| Self::new(schema, f)).collect()) + } + DataType::List(element) => Self::List(Box::new(Self::new(schema, element))), + DataType::Map(entries, _) => Self::Map(Box::new(Self::new(schema, entries))), + _ => Self::Cast, + }; + if result.has_uuid() { + result + } else { + Self::Cast + } + } + + fn has_uuid(&self) -> bool { + match self { + Self::Cast => false, + Self::Uuid => true, + Self::Struct(fields) => fields.iter().any(Self::has_uuid), + Self::List(element) | Self::Map(element) => element.has_uuid(), + } + } + + fn convert( + &self, + array: &ArrayRef, + target: &Field, + visible: Option<&[bool]>, + ) -> Result<ArrayRef> { + let cast = |ty| { + arrow::compute::cast_with_options( + array, + ty, + &arrow::compute::CastOptions { + safe: false, + ..Default::default() + }, + ) + }; + let type_error = || { + DataFusionError::Execution(format!( + "Iceberg field {} expected a container matching {}, got {}", + target.name(), + target.data_type(), + array.data_type() + )) + }; + match self { + Self::Cast => Ok(cast(target.data_type())?), + Self::Uuid => { + // FFI normally supplies Utf8 (possibly dictionary-encoded). Arrow also knows how + // to normalize LargeUtf8 and Utf8View without changing string contents. + let strings = cast(&DataType::Utf8)?; + let strings = strings.as_string::<i32>(); + let mut out = FixedSizeBinaryBuilder::with_capacity(array.len(), 16); + for row in 0..array.len() { + if visible.is_some_and(|v| !v[row]) { + // Invisible required children must remain valid Arrow values. They are + // never written, and must not cause errors by parsing arbitrary bytes. + out.append_value([0; 16])?; + } else if strings.is_null(row) { + out.append_null(); + } else { + let bytes = parse_java_uuid(strings.value(row)).ok_or_else(|| { + DataFusionError::Execution(format!( + "Invalid UUID string in Iceberg field {} at batch row {row}", + target.name() + )) + })?; + out.append_value(bytes)?; + } + } + Ok(Arc::new(out.finish())) + } + Self::Struct(conversions) => { + let source = array + .as_any() + .downcast_ref::<StructArray>() + .ok_or_else(type_error)?; + let DataType::Struct(fields) = target.data_type() else { + unreachable!() + }; + if source.num_columns() != fields.len() { + return Err(DataFusionError::Execution( + "Iceberg struct field count mismatch".into(), + )); + } + let active: Vec<bool> = (0..array.len()) + .map(|i| source.is_valid(i) && visible.is_none_or(|v| v[i])) + .collect(); + let columns = conversions + .iter() + .zip(source.columns()) + .zip(fields) + .map(|((conversion, column), field)| { + conversion.convert(column, field, Some(&active)) + }) + .collect::<Result<Vec<_>>>()?; + Ok(Arc::new(StructArray::try_new( + fields.clone(), + columns, + source.nulls().cloned(), + )?)) + } + Self::List(element) => { + let source = array + .as_any() + .downcast_ref::<ListArray>() + .ok_or_else(type_error)?; + let DataType::List(field) = target.data_type() else { + unreachable!() + }; + let active = child_visibility( + array, + visible, + source.value_offsets(), + source.values().len(), + ); + let values = element.convert(source.values(), field, Some(&active))?; + Ok(Arc::new(ListArray::try_new( + Arc::clone(field), + source.offsets().clone(), + values, + source.nulls().cloned(), + )?)) + } + Self::Map(entries) => { + let source = array + .as_any() + .downcast_ref::<MapArray>() + .ok_or_else(type_error)?; + let DataType::Map(field, ordered) = target.data_type() else { + unreachable!() + }; + let active = child_visibility( + array, + visible, + source.value_offsets(), + source.entries().len(), + ); + let values = entries.convert( + &(Arc::new(source.entries().clone()) as ArrayRef), + field, + Some(&active), + )?; + Ok(Arc::new(MapArray::try_new( + Arc::clone(field), + source.offsets().clone(), + values.as_struct().clone(), + source.nulls().cloned(), + *ordered, + )?)) + } + } + } +} + +fn child_visibility( + array: &ArrayRef, + visible: Option<&[bool]>, + offsets: &[i32], + len: usize, +) -> Vec<bool> { + let mut active = vec![false; len]; + for row in 0..array.len() { + if array.is_valid(row) && visible.is_none_or(|v| v[row]) { + active[offsets[row] as usize..offsets[row + 1] as usize].fill(true); + } + } + active Review Comment: `child_visibility` allocates a `Vec<bool>` the length of the child values for every list and map on every batch. `parquet_support.rs` already has `repeated_visibility`, which handles entries outside a slice and under null parents and returns `None` when nothing is hidden, so the common case allocates nothing. Could it be shared, and the struct `active` mask skipped when there are no nulls and `visible` is `None`? The signatures differ (offsets plus a parent `NullBuffer` against `Option<&[bool]>`), so this needs a check that it fits. ########## native/core/src/execution/operators/iceberg_uuid.rs: ########## @@ -0,0 +1,819 @@ +// 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. + +//! Iceberg UUID strings use Java's `UUID.fromString` semantics, including its permissive +//! five-group fallback. Only Iceberg UUID fields use this conversion; `fixed(16)` stays binary. + +use std::sync::Arc; + +use arrow::array::{ + Array, ArrayRef, AsArray, FixedSizeBinaryBuilder, ListArray, MapArray, StructArray, +}; +use arrow::datatypes::{DataType, Field, SchemaRef}; +use arrow::record_batch::RecordBatch; +use datafusion::error::{DataFusionError, Result}; +use iceberg::spec::{PrimitiveType, Schema, Type}; +use parquet::arrow::PARQUET_FIELD_ID_META_KEY; + +/// Java parses UTF-16 code units, so supplementary-plane digits are deliberately excluded. +/// This table is pinned against Character.digit(char, 16) on JDK 17 and 21 over the entire BMP. +fn java_hex_digit(c: char) -> Option<u64> { + match c { + '0'..='9' => Some(c as u64 - '0' as u64), + 'a'..='f' => Some(c as u64 - 'a' as u64 + 10), + 'A'..='F' => Some(c as u64 - 'A' as u64 + 10), + '\u{ff21}'..='\u{ff26}' => Some(c as u64 - 0xff21 + 10), + '\u{ff41}'..='\u{ff46}' => Some(c as u64 - 0xff41 + 10), + _ => { + const ZEROS: &[u32] = &[ + 0x660, 0x6f0, 0x7c0, 0x966, 0x9e6, 0xa66, 0xae6, 0xb66, 0xbe6, 0xc66, 0xce6, 0xd66, + 0xde6, 0xe50, 0xed0, 0xf20, 0x1040, 0x1090, 0x17e0, 0x1810, 0x1946, 0x19d0, 0x1a80, + 0x1a90, 0x1b50, 0x1bb0, 0x1c40, 0x1c50, 0xa620, 0xa8d0, 0xa900, 0xa9d0, 0xa9f0, + 0xaa50, 0xabf0, 0xff10, + ]; + let value = c as u32; + ZEROS.iter().find_map(|&zero| { + (value >= zero && value < zero + 10).then(|| u64::from(value - zero)) + }) + } + } +} + +fn parse_java_uuid(value: &str) -> Option<[u8; 16]> { + // Avoid the group parser and Unicode handling for canonical ASCII strings. parse_str alone + // also accepts forms Java rejects, so require the canonical dash positions first. + let bytes = value.as_bytes(); + if bytes.len() == 36 && [8, 13, 18, 23].iter().all(|&i| bytes[i] == b'-') { + if let Ok(uuid) = uuid::Uuid::parse_str(value) { + return Some(*uuid.as_bytes()); + } + } + if value.encode_utf16().take(37).count() > 36 { + return None; + } + let mut groups = value.split('-'); + let mut parts = [0u64; 5]; + for part in &mut parts { + let group = groups.next()?; + let digits = group.strip_prefix('+').unwrap_or(group); + if digits.is_empty() { + return None; + } + let mut number = 0u64; + for c in digits.chars() { + number = number.checked_mul(16)?.checked_add(java_hex_digit(c)?)?; + // Long.parseLong rejects signed overflow *before* UUID masks the group width. + if number > i64::MAX as u64 { + return None; + } + } + *part = number; + } + if groups.next().is_some() { + return None; + } + let most = ((parts[0] & 0xffff_ffff) << 32) | ((parts[1] & 0xffff) << 16) | (parts[2] & 0xffff); + let least = ((parts[3] & 0xffff) << 48) | (parts[4] & 0xffff_ffff_ffff); + let mut out = [0u8; 16]; + out[..8].copy_from_slice(&most.to_be_bytes()); + out[8..].copy_from_slice(&least.to_be_bytes()); + Some(out) +} + +/// Constructed once per task. Subtrees without UUID retain the ordinary Arrow cast path. +pub(super) struct UuidAlignment(Vec<Conversion>); + +enum Conversion { + Cast, + Uuid, + Struct(Vec<Conversion>), + List(Box<Conversion>), + Map(Box<Conversion>), +} + +impl UuidAlignment { + pub(super) fn new(schema: &Schema, target: &SchemaRef) -> Self { + Self( + target + .fields() + .iter() + .map(|f| Conversion::new(schema, f)) + .collect(), + ) + } + + pub(super) fn convert(&self, batch: RecordBatch, target: &SchemaRef) -> Result<RecordBatch> { + if batch.num_columns() != self.0.len() { + return Err(DataFusionError::Execution( + "Iceberg write column count mismatch".into(), + )); + } + let columns = batch + .columns() + .iter() + .zip(target.fields()) + .zip(&self.0) + .map(|((array, field), conversion)| conversion.convert(array, field, None)) + .collect::<Result<Vec<_>>>()?; + Ok(RecordBatch::try_new(Arc::clone(target), columns)?) + } +} + +impl Conversion { + fn new(schema: &Schema, field: &Field) -> Self { + if field + .metadata() + .get(PARQUET_FIELD_ID_META_KEY) + .and_then(|id| id.parse().ok()) + .and_then(|id| schema.field_by_id(id)) + .is_some_and(|f| *f.field_type == Type::Primitive(PrimitiveType::Uuid)) + { + return Self::Uuid; + } + let result = match field.data_type() { + DataType::Struct(fields) => { + Self::Struct(fields.iter().map(|f| Self::new(schema, f)).collect()) + } + DataType::List(element) => Self::List(Box::new(Self::new(schema, element))), + DataType::Map(entries, _) => Self::Map(Box::new(Self::new(schema, entries))), + _ => Self::Cast, + }; + if result.has_uuid() { + result + } else { + Self::Cast + } + } + + fn has_uuid(&self) -> bool { + match self { + Self::Cast => false, + Self::Uuid => true, + Self::Struct(fields) => fields.iter().any(Self::has_uuid), + Self::List(element) | Self::Map(element) => element.has_uuid(), + } + } + + fn convert( + &self, + array: &ArrayRef, + target: &Field, + visible: Option<&[bool]>, + ) -> Result<ArrayRef> { + let cast = |ty| { + arrow::compute::cast_with_options( + array, + ty, + &arrow::compute::CastOptions { + safe: false, + ..Default::default() + }, + ) + }; + let type_error = || { + DataFusionError::Execution(format!( + "Iceberg field {} expected a container matching {}, got {}", + target.name(), + target.data_type(), + array.data_type() + )) + }; + match self { + Self::Cast => Ok(cast(target.data_type())?), + Self::Uuid => { + // FFI normally supplies Utf8 (possibly dictionary-encoded). Arrow also knows how + // to normalize LargeUtf8 and Utf8View without changing string contents. + let strings = cast(&DataType::Utf8)?; + let strings = strings.as_string::<i32>(); + let mut out = FixedSizeBinaryBuilder::with_capacity(array.len(), 16); + for row in 0..array.len() { + if visible.is_some_and(|v| !v[row]) { + // Invisible required children must remain valid Arrow values. They are + // never written, and must not cause errors by parsing arbitrary bytes. + out.append_value([0; 16])?; + } else if strings.is_null(row) { + out.append_null(); + } else { + let bytes = parse_java_uuid(strings.value(row)).ok_or_else(|| { + DataFusionError::Execution(format!( + "Invalid UUID string in Iceberg field {} at batch row {row}", Review Comment: `batch row {row}` is an index inside one Comet batch, so a user cannot map it back to their data. Java's `UUID.fromString` error names the offending value. Could this message include the value, truncated, next to the field name? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
