parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3654193866
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -376,39 +459,62 @@ impl RecordBatchTransformer {
let fields: Result<Vec<_>> = projected_iceberg_field_ids
.iter()
.map(|field_id| {
- // Metadata/virtual fields (like _file, _spec_id) don't exist
in the table
- // schema, so build their Arrow field from the metadata column
definition and
- // the pre-computed constant's type.
- if let Some(datum) = constant_fields.get(field_id)
- && let Ok(iceberg_field) = get_metadata_field(*field_id)
- {
- let arrow_type = datum_to_arrow_type_with_ree(datum);
- let arrow_field =
- Field::new(&iceberg_field.name, arrow_type,
!iceberg_field.required)
- .with_metadata(HashMap::from([(
- PARQUET_FIELD_ID_META_KEY.to_string(),
- iceberg_field.id.to_string(),
- )]));
- Ok(Arc::new(arrow_field))
- } else if virtual_fields.contains(field_id) {
+ match constant_fields.get(field_id) {
+ Some(ColumnConstant::Struct(pc))
+ if *field_id == RESERVED_FIELD_ID_PARTITION =>
+ {
+ let struct_type =
DataType::Struct(pc.fields().clone());
+ let nullable = pc.fields().is_empty();
+ let arrow_field = field_with_id(
+ RESERVED_COL_NAME_PARTITION,
+ struct_type,
+ nullable,
+ *field_id,
+ );
+ return Ok(Arc::new(arrow_field));
+ }
+ Some(ColumnConstant::Struct(_)) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ format!("Unexpected struct constant for field id
{field_id}"),
Review Comment:
done
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,68 @@ enum SchemaComparison {
/// Builder for RecordBatchTransformer to improve ergonomics when constructing
with optional parameters.
///
-/// Constant fields are pre-computed for both virtual/metadata fields (like
_file) and
-/// identity-partitioned fields to avoid duplicate work during batch
processing.
+/// All per-file constants (scalar metadata like `_file`, identity partition
values,
+/// and the `_partition` struct) are stored in a single `constant_fields` map
keyed by
+/// field_id. This unified representation (via [`ColumnConstant`]) means the
+/// transformer handles all constant columns through one code path.
#[derive(Debug)]
pub(crate) struct RecordBatchTransformerBuilder {
snapshot_schema: Arc<IcebergSchema>,
projected_iceberg_field_ids: Vec<i32>,
- constant_fields: HashMap<i32, Datum>,
+ constant_fields: HashMap<i32, ColumnConstant>,
virtual_fields: HashSet<i32>,
}
+/// A per-file constant value for a column.
+///
+/// Covers both scalar constants (metadata columns like `_file` and `_spec_id`,
+/// as well as identity partition source fields) and the struct `_partition`
column.
+#[derive(Debug, Clone, PartialEq)]
+pub enum ColumnConstant {
+ /// A scalar constant (e.g., `_file` path, `_spec_id`, identity partition
values).
+ /// The Datum carries both the Iceberg type and the value.
+ Scalar(Datum),
+ /// A struct constant (the `_partition` column). Each child is a primitive
constant
+ /// or null (for partition evolution gaps).
+ Struct(StructConstant),
+}
+
+/// Pre-computed data for a struct constant column.
+#[derive(Debug, Clone, PartialEq)]
+pub struct StructConstant {
Review Comment:
fixed
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,68 @@ enum SchemaComparison {
/// Builder for RecordBatchTransformer to improve ergonomics when constructing
with optional parameters.
///
-/// Constant fields are pre-computed for both virtual/metadata fields (like
_file) and
-/// identity-partitioned fields to avoid duplicate work during batch
processing.
+/// All per-file constants (scalar metadata like `_file`, identity partition
values,
+/// and the `_partition` struct) are stored in a single `constant_fields` map
keyed by
+/// field_id. This unified representation (via [`ColumnConstant`]) means the
+/// transformer handles all constant columns through one code path.
#[derive(Debug)]
pub(crate) struct RecordBatchTransformerBuilder {
snapshot_schema: Arc<IcebergSchema>,
projected_iceberg_field_ids: Vec<i32>,
- constant_fields: HashMap<i32, Datum>,
+ constant_fields: HashMap<i32, ColumnConstant>,
virtual_fields: HashSet<i32>,
}
+/// A per-file constant value for a column.
+///
+/// Covers both scalar constants (metadata columns like `_file` and `_spec_id`,
+/// as well as identity partition source fields) and the struct `_partition`
column.
+#[derive(Debug, Clone, PartialEq)]
+pub enum ColumnConstant {
Review Comment:
fixed
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,101 @@
+// 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::HashSet;
+
+use crate::spec::{NestedField, NestedFieldRef, PartitionSpec, Schema,
StructType, Transform};
+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 behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Void and unknown transform fields are skipped.
+/// - Fields are deduplicated by field_id — each unique field_id appears
exactly once.
+///
+/// # 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>(
Review Comment:
added unit tests -
single spec identity,
year transform,
unpartitioned,
multiple fields sorted by id,
newer name precedence, void replaced by older non-void,
dropped source column skipped, and partition evolution adding new fields
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,101 @@
+// 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::HashSet;
+
+use crate::spec::{NestedField, NestedFieldRef, PartitionSpec, Schema,
StructType, Transform};
+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 behavior:
+/// - Specs are sorted by spec_id in descending order (newer specs first), so
newer field
+/// names take precedence when deduplicating by field_id.
+/// - Void and unknown transform fields are skipped.
+/// - Fields are deduplicated by field_id — each unique field_id appears
exactly once.
+///
+/// # 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 seen_field_ids = HashSet::new();
+ let mut struct_fields: Vec<NestedFieldRef> = Vec::new();
+
+ // Sort specs by spec_id descending (newer first) to match Java's behavior:
+ // newer field names take precedence when deduplicating by field_id.
+ let mut specs: Vec<&PartitionSpec> = partition_specs.collect();
+ specs.sort_by_key(|s| Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
Review Comment:
Great point @ctty. This did not match the java version. Redid this to match
java's `buildPartitionProjectionType`. so that when a newer spec marks a field
Void 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. Introduced
`all_active_field_ids` helper matching Java's `allActiveFieldIds`.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -723,6 +845,89 @@ impl RecordBatchTransformer {
create_primitive_array_repeated(target_type, prim_lit, num_rows)
}
}
+
+ fn create_struct_column(
+ fields: &Fields,
+ child_values: &[Option<PrimitiveLiteral>],
+ num_rows: usize,
+ ) -> Result<ArrayRef> {
+ if fields.is_empty() {
+ let nulls = arrow_buffer::NullBuffer::new_null(num_rows);
+ return Ok(Arc::new(StructArray::new_empty_fields(
+ num_rows,
+ Some(nulls),
+ )));
+ }
+
+ let child_arrays: Vec<ArrayRef> = fields
+ .iter()
+ .zip(child_values.iter())
+ .map(|(field, value)| {
+ create_primitive_array_repeated(field.data_type(), value,
num_rows)
+ })
+ .collect::<Result<_>>()?;
+
+ Ok(Arc::new(StructArray::try_new(
+ fields.clone(),
+ child_arrays,
+ None,
+ )?))
+ }
+}
+
+/// Builds a [`StructConstant`] from the unified partition type and a file's
+/// partition spec/data.
+///
+/// If `unified_partition_type` has no fields (unpartitioned table), returns
an empty constant
+/// that renders as a null struct column.
+///
+/// For each field in the unified partition type:
+/// - If it corresponds to a field in this file's partition spec, use the
value from partition_data
+/// - Otherwise (partition evolution), use null
+pub fn build_partition_column_constant(
Review Comment:
fixed
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -249,6 +322,18 @@ impl RecordBatchTransformerBuilder {
self
}
+ /// Set a pre-computed _partition column constant directly.
+ pub(crate) fn with_partition_column_precomputed(
Review Comment:
fixed
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -501,39 +607,50 @@ 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).
- if let Some(datum) = constant_fields.get(field_id) {
- let is_metadata_field =
get_metadata_field(*field_id).is_ok();
- let present_in_file =
field_id_to_source_schema_map.contains_key(field_id);
-
- if is_metadata_field || !present_in_file {
- let arrow_type = if is_metadata_field {
- datum_to_arrow_type_with_ree(datum)
- } else {
- field_id_to_mapped_schema_map
- .get(field_id)
- .ok_or(Error::new(
- ErrorKind::Unexpected,
- "could not find field in schema",
- ))?
- .0
- .data_type()
- .clone()
- };
-
- return Ok(ColumnSource::Add {
- value: Some(datum.literal().clone()),
- target_type: arrow_type,
+ match constant_fields.get(field_id) {
+ Some(ColumnConstant::Struct(pc))
+ if *field_id == RESERVED_FIELD_ID_PARTITION =>
+ {
+ return Ok(ColumnSource::AddStructConstant {
+ fields: pc.fields().clone(),
+ child_values: pc.child_values().to_vec(),
});
}
+ Some(ColumnConstant::Struct(_)) => {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ format!("Unexpected struct constant for field id
{field_id}"),
Review Comment:
changed
--
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]
