Copilot commented on code in PR #2773:
URL: https://github.com/apache/iceberg-rust/pull/2773#discussion_r3845029527
##########
crates/iceberg/src/spec/schema/mod.rs:
##########
@@ -1475,4 +1563,244 @@ table {
.is_err()
);
}
+
+ #[test]
+ fn test_unknown_type_deserialization_rejects_non_null_default() {
+ let schema_json = serde_json::json!({
+ "type": "struct",
+ "schema-id": 1,
+ "fields": [
+ {
+ "id": 1,
+ "name": "empty",
+ "required": false,
+ "type": "unknown",
+ "initial-default": 1
+ }
+ ]
+ });
+
+ let error = serde_json::from_value::<Schema>(schema_json).unwrap_err();
+ assert!(
+ error
+ .to_string()
+ .contains("did not match any variant of untagged enum
SchemaEnum"),
Review Comment:
This test assertion depends on a serde/internal error string (`"did not
match any variant of untagged enum SchemaEnum"`), which is brittle across serde
versions and unrelated refactors. It would be more stable to assert on the
domain error message you control (e.g., the specific “unknown defaults must be
null” validation message) so the test fails only when the intended behavior
changes.
##########
crates/iceberg/src/arrow/reader/projection.rs:
##########
@@ -460,13 +441,20 @@ pub(super) fn add_fallback_field_ids_to_arrow_schema(
"Schema already has field IDs"
);
+ let omitted_field_ids: HashSet<i32> = iceberg_schema
+ .as_struct()
+ .fields()
+ .iter()
+ .filter(|field| !type_has_parquet_physical_field(&field.field_type))
+ .map(|field| field.id)
+ .collect();
+ let mut fallback_field_ids = (1_i32..).filter(|field_id|
!omitted_field_ids.contains(field_id));
let fields_with_fallback_ids: Vec<_> = arrow_schema
.fields()
.iter()
- .enumerate()
- .map(|(pos, field)| {
+ .map(|field| {
let mut metadata = field.metadata().clone();
- let field_id = (pos + 1) as i32; // 1-indexed for Java
compatibility
+ let field_id = fallback_field_ids.next().unwrap();
Review Comment:
`add_fallback_field_ids_to_arrow_schema` is mixing *real Iceberg field IDs*
(`field.id`) with the *ordinal fallback IDs* (`pos+1`) used for files without
embedded IDs. If a table schema uses non-sequential/non-1-based field IDs
(common after schema evolution), this will fail to skip the correct ordinal
slots for omitted unknown fields and can assign wrong fallback IDs to
subsequent physical columns. A concrete fix is to compute the omitted *ordinal
positions* (index + 1 of top-level fields that have no Parquet physical field),
and skip those ordinals when generating fallback IDs, rather than skipping the
actual `field.id` values.
##########
crates/iceberg/src/writer/file_writer/parquet_writer.rs:
##########
@@ -177,6 +183,168 @@ impl FileWriterBuilder for ParquetWriterBuilder {
}
}
+fn field_id(field: &FieldRef) -> Option<&str> {
+ field
+ .metadata()
+ .get(PARQUET_FIELD_ID_META_KEY)
+ .map(String::as_str)
+}
+
+fn find_field_index(
+ fields: &Fields,
+ target: &FieldRef,
+ match_mode: FieldMatchMode,
+) -> Option<usize> {
+ match match_mode {
+ FieldMatchMode::Id => field_id(target).and_then(|target_id| {
+ fields
+ .iter()
+ .position(|field| field_id(field) == Some(target_id))
+ }),
+ FieldMatchMode::Name => fields
+ .iter()
+ .position(|field| field.name() == target.name()),
+ }
+}
+
+fn project_array_for_parquet(
+ array: &ArrayRef,
+ target: &FieldRef,
+ match_mode: FieldMatchMode,
+) -> Result<ArrayRef> {
+ if array.data_type() == target.data_type() {
+ return Ok(array.clone());
+ }
+
+ match target.data_type() {
+ DataType::Struct(target_fields) => {
+ let source = array
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Expected struct array for Parquet field {}, got
{}",
+ target.name(),
+ array.data_type()
+ ),
+ )
+ })?;
+ let columns = target_fields
+ .iter()
+ .map(|target_field| {
+ let index = find_field_index(source.fields(),
target_field, match_mode)
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Field {} is missing from struct array for
Parquet write",
+ target_field.name()
+ ),
+ )
+ })?;
+ project_array_for_parquet(source.column(index),
target_field, match_mode)
+ })
+ .collect::<Result<Vec<_>>>()?;
+ Ok(Arc::new(StructArray::try_new_with_length(
+ target_fields.clone(),
+ columns,
+ source.nulls().cloned(),
+ source.len(),
+ )?))
+ }
+ DataType::List(target_element) => {
+ let source =
array.as_any().downcast_ref::<ListArray>().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Expected list array for Parquet field {}, got {}",
+ target.name(),
+ array.data_type()
+ ),
+ )
+ })?;
+ let values = project_array_for_parquet(source.values(),
target_element, match_mode)?;
+ Ok(Arc::new(ListArray::try_new(
+ target_element.clone(),
+ source.offsets().clone(),
+ values,
+ source.nulls().cloned(),
+ )?))
+ }
+ DataType::Map(target_entries, ordered) => {
+ let source =
array.as_any().downcast_ref::<MapArray>().ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Expected map array for Parquet field {}, got {}",
+ target.name(),
+ array.data_type()
+ ),
+ )
+ })?;
+ let source_entries: ArrayRef = Arc::new(source.entries().clone());
+ let entries = project_array_for_parquet(&source_entries,
target_entries, match_mode)?;
+ let entries = entries
+ .as_any()
+ .downcast_ref::<StructArray>()
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Projected Parquet map entries are not a struct array",
+ )
+ })?
+ .clone();
+ Ok(Arc::new(MapArray::try_new(
+ target_entries.clone(),
+ source.offsets().clone(),
+ entries,
+ source.nulls().cloned(),
+ *ordered,
+ )?))
+ }
+ _ => Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Cannot project Arrow type {} to {} for Parquet field {}",
+ array.data_type(),
+ target.data_type(),
+ target.name()
+ ),
+ )),
+ }
+}
+
+fn project_batch_for_parquet(
+ batch: &RecordBatch,
+ target_schema: ArrowSchemaRef,
+ match_mode: FieldMatchMode,
+) -> Result<RecordBatch> {
+ let source_schema = batch.schema();
+ let columns = target_schema
+ .fields()
+ .iter()
+ .map(|target_field| {
+ let index = find_field_index(source_schema.fields(), target_field,
match_mode)
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "Field {} is missing from record batch for Parquet
write",
+ target_field.name()
+ ),
+ )
+ })?;
+ project_array_for_parquet(batch.column(index), target_field,
match_mode)
+ })
+ .collect::<Result<Vec<_>>>()?;
Review Comment:
`project_batch_for_parquet` always walks the target schema and potentially
reconstructs arrays, even when the incoming batch schema already matches
`target_schema` (common case when no unknown fields are present or the batch
was produced using the same schema). Consider adding a fast-path early return
when the schemas are equal (or when no projection is needed), to avoid
per-batch overhead in write-heavy workloads.
##########
crates/iceberg/src/spec/schema/mod.rs:
##########
@@ -190,6 +194,79 @@ impl SchemaBuilder {
Ok(schema)
}
+ fn validate_unknown_type_field(field: &NestedFieldRef) -> Result<()> {
+ ensure_data_valid!(
+ !field
+ .initial_default
+ .iter()
+ .chain(field.write_default.iter())
+ .any(|default| {
+ Self::default_contains_non_null_unknown(default,
&field.field_type)
+ }),
+ "Field {} cannot have non-null defaults because unknown type
requires null defaults",
+ field.name
+ );
+
+ match field.field_type.as_ref() {
+ Type::Primitive(PrimitiveType::Unknown) => {
+ ensure_data_valid!(
+ !field.required,
+ "Field {} cannot be required because unknown type must be
optional",
+ field.name
+ );
+ }
+ Type::Struct(struct_type) => {
+ for nested_field in struct_type.fields() {
+ Self::validate_unknown_type_field(nested_field)?;
+ }
+ }
+ Type::List(list_type) => {
+ Self::validate_unknown_type_field(&list_type.element_field)?;
+ }
+ Type::Map(map_type) => {
+ Self::validate_unknown_type_field(&map_type.key_field)?;
+ Self::validate_unknown_type_field(&map_type.value_field)?;
+ }
+ Type::Primitive(_) | Type::Variant(_) => {}
+ }
+
+ Ok(())
+ }
+
+ fn default_contains_non_null_unknown(default: &Literal, field_type: &Type)
-> bool {
+ match (default, field_type) {
+ (_, Type::Primitive(PrimitiveType::Unknown)) => true,
+ (Literal::Struct(value), Type::Struct(struct_type)) => value
+ .iter()
+ .zip(struct_type.fields())
+ .any(|(value, field)| {
+ value.is_some_and(|value| {
+ Self::default_contains_non_null_unknown(value,
&field.field_type)
+ })
+ }),
+ (Literal::List(values), Type::List(list_type)) =>
values.iter().any(|value| {
+ value.as_ref().is_some_and(|value| {
+ Self::default_contains_non_null_unknown(
+ value,
+ &list_type.element_field.field_type,
+ )
+ })
+ }),
+ (Literal::Map(map), Type::Map(map_type)) => {
+ map.clone().into_iter().any(|(key, value)| {
+ Self::default_contains_non_null_unknown(&key,
&map_type.key_field.field_type)
+ || value.as_ref().is_some_and(|value| {
+ Self::default_contains_non_null_unknown(
+ value,
+ &map_type.value_field.field_type,
+ )
+ })
+ })
+ }
Review Comment:
`map.clone().into_iter()` clones the entire map value just to iterate, which
is avoidable if the map literal type supports borrowing iteration (e.g.,
`iter()`). Since this runs during schema building/validation it may be
acceptable, but if schemas are built frequently or defaults are large,
iterating by reference would reduce allocations and work.
--
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]
