CTTY commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3669409024
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -501,39 +608,54 @@ impl RecordBatchTransformer {
projected_iceberg_field_ids
.iter()
.map(|field_id| {
- // Check if this is a constant field (metadata/virtual or
identity-partitioned).
- //
- // Metadata/virtual fields (like _file, _spec_id) never exist
in the data file,
- // so they always use their pre-computed constant value.
- //
- // For identity-partitioned fields, the Iceberg spec's "Column
Projection" rules
- // only apply to "field ids which are not present in a data
file". When the column
- // IS present in the Parquet file, it must be read from the
file; the partition
- // metadata constant is only a fallback for when the column is
absent (e.g. add_files).
Review Comment:
this comment still holds true, can we keep this?
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,347 @@
+// 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.
+
+//! Partition type utilities for Iceberg tables.
+
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::{
+ NestedField, NestedFieldRef, PartitionSpec, Schema, StructType, Transform,
Type,
+};
+use crate::{Error, ErrorKind, Result};
+
+/// Computes the unified partition type across all partition specs in the
table.
+///
+/// This is equivalent to Java's `Partitioning.partitionType(table)`. The
result is a
+/// StructType containing all partition fields ever used across all specs,
enabling correct
+/// representation of the `_partition` metadata column when partition
evolution has occurred.
+///
+/// Matches Java's `buildPartitionProjectionType` behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Unknown transforms cause an error.
+/// - Fields whose source column was dropped from the schema are skipped.
+/// - When a newer spec marks a field as Void (dropped) but an older spec has
it with a
+/// real transform, the older spec's type is preserved while the newer
spec's name is kept.
+/// - Fields are deduplicated by field_id; each unique field_id appears
exactly once.
+/// - Output fields are sorted by field_id ascending.
+///
+/// # Arguments
+/// * `partition_specs` - Iterator over all partition specs in the table
+/// * `schema` - The current table schema (needed to determine result types of
transforms)
+pub fn compute_unified_partition_type<'a>(
+ partition_specs: impl Iterator<Item = &'a PartitionSpec>,
+ schema: &Schema,
+) -> Result<StructType> {
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ let active_field_ids = all_active_field_ids(specs.iter().copied(), schema);
+
+ let mut field_map: HashMap<i32, &crate::spec::PartitionField> =
HashMap::new();
+ let mut type_map: HashMap<i32, Type> = HashMap::new();
+ let mut name_map: HashMap<i32, String> = HashMap::new();
+
+ for spec in &specs {
+ for field in spec.fields() {
+ let field_id = field.field_id;
+
+ if !active_field_ids.contains(&field_id) {
+ continue;
+ }
+
+ if matches!(field.transform, Transform::Unknown) {
Review Comment:
This check should be moved before the continue above, otherwise it may be
skipped
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,347 @@
+// 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.
+
+//! Partition type utilities for Iceberg tables.
+
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::{
+ NestedField, NestedFieldRef, PartitionSpec, Schema, StructType, Transform,
Type,
+};
+use crate::{Error, ErrorKind, Result};
+
+/// Computes the unified partition type across all partition specs in the
table.
+///
+/// This is equivalent to Java's `Partitioning.partitionType(table)`. The
result is a
+/// StructType containing all partition fields ever used across all specs,
enabling correct
+/// representation of the `_partition` metadata column when partition
evolution has occurred.
+///
+/// Matches Java's `buildPartitionProjectionType` behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Unknown transforms cause an error.
+/// - Fields whose source column was dropped from the schema are skipped.
+/// - When a newer spec marks a field as Void (dropped) but an older spec has
it with a
+/// real transform, the older spec's type is preserved while the newer
spec's name is kept.
+/// - Fields are deduplicated by field_id; each unique field_id appears
exactly once.
+/// - Output fields are sorted by field_id ascending.
+///
+/// # Arguments
+/// * `partition_specs` - Iterator over all partition specs in the table
+/// * `schema` - The current table schema (needed to determine result types of
transforms)
+pub fn compute_unified_partition_type<'a>(
+ partition_specs: impl Iterator<Item = &'a PartitionSpec>,
+ schema: &Schema,
+) -> Result<StructType> {
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ let active_field_ids = all_active_field_ids(specs.iter().copied(), schema);
+
+ let mut field_map: HashMap<i32, &crate::spec::PartitionField> =
HashMap::new();
+ let mut type_map: HashMap<i32, Type> = HashMap::new();
+ let mut name_map: HashMap<i32, String> = HashMap::new();
+
+ for spec in &specs {
+ for field in spec.fields() {
+ let field_id = field.field_id;
+
+ if !active_field_ids.contains(&field_id) {
+ continue;
+ }
+
+ if matches!(field.transform, Transform::Unknown) {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
Review Comment:
I'd say it's data invalid error. We can't support unknown transform since we
don't know the result type
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -311,6 +312,24 @@ impl FileScanTaskReader {
record_batch_transformer_builder.with_virtual_field(RESERVED_FIELD_ID_POS);
}
+ // Add the _partition metadata struct column if it's in the projected
fields.
+ // Computed lazily here at read time from the unified partition type +
task's spec + data.
+ if task
+ .project_field_ids()
+ .contains(&RESERVED_FIELD_ID_PARTITION)
+ && let Some(unified_type) = &task.unified_partition_type
+ {
+ let (spec, partition_data) = match (&task.partition_spec,
&task.partition) {
+ (Some(spec), Some(data)) => (spec.clone(), data.clone()),
+ // Unpartitioned table or missing spec:
build_partition_constant
+ // handles the empty unified_type case with an early return.
+ _ => (Arc::new(PartitionSpec::unpartition_spec()),
Struct::empty()),
Review Comment:
We should fail if !unified_type.fields().is_empty()
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,347 @@
+// 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.
+
+//! Partition type utilities for Iceberg tables.
+
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::{
+ NestedField, NestedFieldRef, PartitionSpec, Schema, StructType, Transform,
Type,
+};
+use crate::{Error, ErrorKind, Result};
+
+/// Computes the unified partition type across all partition specs in the
table.
+///
+/// This is equivalent to Java's `Partitioning.partitionType(table)`. The
result is a
+/// StructType containing all partition fields ever used across all specs,
enabling correct
+/// representation of the `_partition` metadata column when partition
evolution has occurred.
+///
+/// Matches Java's `buildPartitionProjectionType` behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Unknown transforms cause an error.
+/// - Fields whose source column was dropped from the schema are skipped.
+/// - When a newer spec marks a field as Void (dropped) but an older spec has
it with a
+/// real transform, the older spec's type is preserved while the newer
spec's name is kept.
+/// - Fields are deduplicated by field_id; each unique field_id appears
exactly once.
+/// - Output fields are sorted by field_id ascending.
+///
+/// # Arguments
+/// * `partition_specs` - Iterator over all partition specs in the table
+/// * `schema` - The current table schema (needed to determine result types of
transforms)
+pub fn compute_unified_partition_type<'a>(
+ partition_specs: impl Iterator<Item = &'a PartitionSpec>,
+ schema: &Schema,
+) -> Result<StructType> {
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ let active_field_ids = all_active_field_ids(specs.iter().copied(), schema);
+
+ let mut field_map: HashMap<i32, &crate::spec::PartitionField> =
HashMap::new();
Review Comment:
nit: we can just import this
```suggestion
let mut field_map: HashMap<i32, &PartitionField> = HashMap::new();
```
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,347 @@
+// 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.
+
+//! Partition type utilities for Iceberg tables.
+
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::{
+ NestedField, NestedFieldRef, PartitionSpec, Schema, StructType, Transform,
Type,
+};
+use crate::{Error, ErrorKind, Result};
+
+/// Computes the unified partition type across all partition specs in the
table.
+///
+/// This is equivalent to Java's `Partitioning.partitionType(table)`. The
result is a
+/// StructType containing all partition fields ever used across all specs,
enabling correct
+/// representation of the `_partition` metadata column when partition
evolution has occurred.
+///
+/// Matches Java's `buildPartitionProjectionType` behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Unknown transforms cause an error.
+/// - Fields whose source column was dropped from the schema are skipped.
+/// - When a newer spec marks a field as Void (dropped) but an older spec has
it with a
+/// real transform, the older spec's type is preserved while the newer
spec's name is kept.
+/// - Fields are deduplicated by field_id; each unique field_id appears
exactly once.
+/// - Output fields are sorted by field_id ascending.
+///
+/// # Arguments
+/// * `partition_specs` - Iterator over all partition specs in the table
+/// * `schema` - The current table schema (needed to determine result types of
transforms)
+pub fn compute_unified_partition_type<'a>(
+ partition_specs: impl Iterator<Item = &'a PartitionSpec>,
+ schema: &Schema,
+) -> Result<StructType> {
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ let active_field_ids = all_active_field_ids(specs.iter().copied(), schema);
+
+ let mut field_map: HashMap<i32, &crate::spec::PartitionField> =
HashMap::new();
+ let mut type_map: HashMap<i32, Type> = HashMap::new();
+ let mut name_map: HashMap<i32, String> = HashMap::new();
+
+ for spec in &specs {
+ for field in spec.fields() {
+ let field_id = field.field_id;
+
+ if !active_field_ids.contains(&field_id) {
+ continue;
+ }
+
+ if matches!(field.transform, Transform::Unknown) {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ format!(
+ "Partition field '{}' uses an unknown transform that
is not \
+ supported by this version of iceberg-rust",
+ field.name
+ ),
+ ));
+ }
+
+ let source_field = match schema.field_by_id(field.source_id) {
+ Some(f) => f,
+ None => continue,
+ };
+
+ match field_map.get(&field_id) {
+ None => {
+ let res_type =
field.transform.result_type(&source_field.field_type)?;
+ field_map.insert(field_id, field);
+ type_map.insert(field_id, res_type);
+ name_map.insert(field_id, field.name.clone());
+ }
+ Some(existing) => {
+ if is_void_transform(existing) &&
!is_void_transform(field) {
Review Comment:
Java does a compatibility check before this to compare the existing field
and the new field, and we need this because V1 doesn't ensure the field id is
unique across specs:
https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/Partitioning.java#L295
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,347 @@
+// 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.
+
+//! Partition type utilities for Iceberg tables.
+
+use std::cmp::Reverse;
+use std::collections::{HashMap, HashSet};
+
+use crate::spec::{
+ NestedField, NestedFieldRef, PartitionSpec, Schema, StructType, Transform,
Type,
+};
+use crate::{Error, ErrorKind, Result};
+
+/// Computes the unified partition type across all partition specs in the
table.
+///
+/// This is equivalent to Java's `Partitioning.partitionType(table)`. The
result is a
+/// StructType containing all partition fields ever used across all specs,
enabling correct
+/// representation of the `_partition` metadata column when partition
evolution has occurred.
+///
+/// Matches Java's `buildPartitionProjectionType` behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Unknown transforms cause an error.
+/// - Fields whose source column was dropped from the schema are skipped.
+/// - When a newer spec marks a field as Void (dropped) but an older spec has
it with a
+/// real transform, the older spec's type is preserved while the newer
spec's name is kept.
+/// - Fields are deduplicated by field_id; each unique field_id appears
exactly once.
+/// - Output fields are sorted by field_id ascending.
+///
+/// # Arguments
+/// * `partition_specs` - Iterator over all partition specs in the table
+/// * `schema` - The current table schema (needed to determine result types of
transforms)
+pub fn compute_unified_partition_type<'a>(
+ partition_specs: impl Iterator<Item = &'a PartitionSpec>,
+ schema: &Schema,
+) -> Result<StructType> {
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ let active_field_ids = all_active_field_ids(specs.iter().copied(), schema);
+
+ let mut field_map: HashMap<i32, &crate::spec::PartitionField> =
HashMap::new();
+ let mut type_map: HashMap<i32, Type> = HashMap::new();
+ let mut name_map: HashMap<i32, String> = HashMap::new();
+
+ for spec in &specs {
+ for field in spec.fields() {
+ let field_id = field.field_id;
+
+ if !active_field_ids.contains(&field_id) {
+ continue;
+ }
+
+ if matches!(field.transform, Transform::Unknown) {
+ return Err(Error::new(
+ ErrorKind::FeatureUnsupported,
+ format!(
+ "Partition field '{}' uses an unknown transform that
is not \
+ supported by this version of iceberg-rust",
+ field.name
+ ),
+ ));
+ }
+
+ let source_field = match schema.field_by_id(field.source_id) {
+ Some(f) => f,
+ None => continue,
+ };
+
+ match field_map.get(&field_id) {
+ None => {
+ let res_type =
field.transform.result_type(&source_field.field_type)?;
+ field_map.insert(field_id, field);
+ type_map.insert(field_id, res_type);
+ name_map.insert(field_id, field.name.clone());
+ }
+ Some(existing) => {
+ if is_void_transform(existing) &&
!is_void_transform(field) {
+ let res_type =
field.transform.result_type(&source_field.field_type)?;
+ field_map.insert(field_id, field);
+ type_map.insert(field_id, res_type);
+ }
+ }
+ }
+ }
+ }
+
+ let mut field_ids: Vec<i32> = field_map.keys().copied().collect();
+ field_ids.sort();
+
+ let struct_fields: Vec<NestedFieldRef> = field_ids
+ .into_iter()
+ .map(|fid| {
+ let name = &name_map[&fid];
+ let ty = type_map.remove(&fid).unwrap();
Review Comment:
I understand that this is unlikely to fail due to the way it was
constructed. But I still think we should not unwrap here, how about the below?
```rust
let struct_fields = field_ids
.into_iter()
.map(|fid| -> Result<NestedFieldRef> {
let name = name_map.get(&fid).ok_or_else(|| {
Error::new(
ErrorKind::Unexpected,
format!("Missing name for partition field {fid}"),
)
})?;
let ty = type_map.remove(&fid).ok_or_else(|| {
Error::new(
ErrorKind::Unexpected,
format!("Missing type for partition field {fid}"),
)
})?;
Ok(NestedField::optional(fid, name, ty).into())
})
.collect::<Result<Vec<_>>>()?;
```
--
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]
