You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "parthchandra (via GitHub)" <gi...@apache.org> on 2026/06/18 00:34:06 UTC
[PR] feat: add support for _partition metadata column [iceberg-rust]
parthchandra opened a new pull request, #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668
## Which issue does this PR close?
- part of #https://github.com/apache/iceberg-rust/issues/2607.
## What changes are included in this PR?
Implements the _partition metadata column for table scans. This is a struct column whose type is the union of all partition fields across all partition specs (handling partition evolution). Each row gets the
partition values for its data file.
- Adds compute_unified_partition_type() to compute the union of partition fields across all specs (equivalent to Java's Partitioning.partitionType())
- Adds PartitionColumnConstant and build_partition_column_constant() for pre-computing the struct values per file
- Adds ColumnSource::AddStructConstant variant to RecordBatchTransformer for materializing struct columns
- Threads the unified partition type through scan planning and populates the constant in FileScanTask
- Pipeline detects RESERVED_FIELD_ID_PARTITION in projected fields and injects the struct constant
## Are these changes tested?
Because we do not have write support yet, I made the corresponding change to comet and then tested by adding tests in Comet which uses iceberg-java to write files and then iceberg-rust to read them back.
https://github.com/parthchandra/datafusion-comet/blob/iceberg-metadata-columns/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3448373215
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -116,6 +125,26 @@ impl ManifestEntryContext {
)
.await;
+ // Compute the _partition struct constant if the unified partition type is available
+ let partition_column_constant =
+ if let Some(ref unified_partition_type) = self.unified_partition_type {
+ let partition_spec = self
+ .table_metadata
+ .partition_spec_by_id(self.partition_spec_id);
+ if let Some(spec) = partition_spec {
+ let constant = build_partition_column_constant(
+ unified_partition_type,
+ spec,
+ &self.manifest_entry.data_file.partition,
+ )?;
+ Some(Arc::new(constant))
+ } else {
+ None
+ }
+ } else {
+ None
+ };
Review Comment:
I'm not sure this is a good idea to calculate the partition column in the plan/scan phase and it adds build_partition_column_constant dep from record_batch_transformer' mod into the scan side.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -116,6 +117,21 @@ pub struct FileScanTask {
#[builder(default)]
pub name_mapping: Option<Arc<NameMapping>>,
+ /// Pre-computed constant for the `_partition` metadata struct column.
+ /// Populated by scan planning (or externally by Comet's planner) when
+ /// `RESERVED_FIELD_ID_PARTITION` is in the projected field IDs.
+ ///
+ /// NOTE: This field is not serializable. FileScanTask cannot be
+ /// serialized/deserialized when this is populated. This is consistent with
+ /// partition_spec and name_mapping above, and is acceptable because Comet
+ /// builds tasks in-process (no serialization round-trip).
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[serde(serialize_with = "serialize_not_implemented")]
+ #[serde(deserialize_with = "deserialize_not_implemented")]
+ #[builder(default)]
+ pub partition_column_constant: Option<Arc<PartitionColumnConstant>>,
Review Comment:
Like the comment in https://github.com/apache/iceberg-rust/pull/2668/changes#r3448373215, I think it's better to pass the unified partition type here rather than passing the actual unified partition value. It would be easier for comet to pooling the type rather than the actual value.
BTW, this would be unnecessary if we can rebuild/access the table/metadata when reading on the executor side. Java archives this by SerializableTable and with Spark's broadcast. We don't have similar thing on the rust yet.
##########
crates/iceberg/src/metadata_columns.rs:
##########
@@ -379,6 +381,52 @@ pub fn partition_field(partition_fields: Vec<NestedFieldRef>) -> NestedFieldRef
)
}
+/// 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.
+///
+/// For each spec, the partition fields are derived from the transform applied to the source
+/// column. Fields are deduplicated by field_id - each unique field_id appears exactly once
+/// in the result.
+///
+/// # 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:
nit: it's a bit of odd that this function is added in metadata_column.rs. Do you think it's a good idea to create partitioning.rs and put this function into partitioning.rs?
##########
crates/iceberg/src/metadata_columns.rs:
##########
@@ -379,6 +381,52 @@ pub fn partition_field(partition_fields: Vec<NestedFieldRef>) -> NestedFieldRef
)
}
+/// 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.
+///
+/// For each spec, the partition fields are derived from the transform applied to the source
+/// column. Fields are deduplicated by field_id - each unique field_id appears exactly once
+/// in the result.
+///
+/// # 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 = std::collections::HashSet::new();
+ let mut struct_fields: Vec<NestedFieldRef> = Vec::new();
+
+ for spec in partition_specs {
Review Comment:
I notice some inconsistent with java's impl:
1. unknown specs are rejected in java.
2. specs are sorted by spec id first(in reverse order), which means newer partition spec's field name will be picked fist.
3. V1 table's void transform field is also handled in java: the partition field that was dropped later.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -140,6 +154,13 @@ pub(crate) enum ColumnSource {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
+
+ // A struct column where each child is a constant primitive value.
+ // Used for the _partition metadata column.
+ AddStructConstant {
Review Comment:
hmmm. why this is not added to the constant fields? I think java's impl simply add partition's field constant map with a struct projection to mapping the task's partition data into the unified type?
##########
crates/iceberg/src/arrow/value.rs:
##########
@@ -909,6 +909,38 @@ pub(crate) fn create_primitive_array_repeated(
let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
Arc::new(BinaryArray::from_opt_vec(vals))
}
+ (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
+ Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows]))
+ }
+ (DataType::LargeBinary, None) => {
+ let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(LargeBinaryArray::from_opt_vec(vals))
+ }
+ (DataType::FixedSizeBinary(len), Some(PrimitiveLiteral::Binary(value))) => {
+ let repeated: Vec<&[u8]> = vec![value.as_slice(); num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_iter(repeated.into_iter()).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
+ (DataType::FixedSizeBinary(len), None) => {
+ let repeated: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(repeated.into_iter(), *len).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create null FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
Review Comment:
these seems not related?
Our internal integration also shows iceberg mapping Iceberg's binary to Arrow's LargeBinary though, which should also be updated.
##########
crates/iceberg/src/spec/schema/_serde.rs:
##########
@@ -100,14 +105,19 @@ impl TryFrom<SchemaV1> for Schema {
}
}
+// Note: alias_to_id is intentionally excluded from serialization.
+// Per Iceberg spec, aliases are not part of the JSON schema representation.
+// They must be reconstructed from external sources after deserialization.
impl From<Schema> for SchemaV2 {
fn from(value: Schema) -> Self {
SchemaV2 {
schema_id: value.schema_id,
identifier_field_ids: if value.identifier_field_ids.is_empty() {
None
} else {
- Some(value.identifier_field_ids.into_iter().collect())
+ let mut ids: Vec<i32> = value.identifier_field_ids.into_iter().collect();
+ ids.sort_unstable();
+ Some(ids)
},
Review Comment:
the changes in this file seem unrelated? And I don't think the spec requires sorting identifier field ids.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4783305963
> Comet generates the FileScanTasks from Iceberg Java. Does Iceberg Java provide `unified_partition_type`, otherwise we need to be sure Comet can generate these if you want Comet to work with this.
Good point! Iceberg Java doesn't provide unified_partition_type on individual FileScanTasks — Java computes it at a higher level via [Partitioning.partitionType(table) ](https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/Partitioning.java#L240-L244)
We can do the equivalent in Comet - https://github.com/parthchandra/datafusion-comet/blob/iceberg-metadata-columns/native/core/src/execution/planner.rs#L3409-L3440
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3457615535
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,80 @@
+// 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 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 transform fields (dropped partition columns) 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 = std::collections::HashSet::new();
+ let mut struct_fields: Vec<NestedFieldRef> = Vec::new();
Review Comment:
nit: should unknown transform should be handled too?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5097138538
Hi @parthchandra , looks like the code is not updated yet. Maybe you have commits that are not pushed to the remote?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4937809487
Comet (draft) PR with @hsiang-c 's PRs included. CI is green. https://github.com/apache/datafusion-comet/pull/4752
(branch with all the PRs merged - https://github.com/parthchandra/iceberg-rust/tree/metadata-columns-unified )
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3628177500
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -300,6 +303,20 @@ impl<'a> TableScanBuilder<'a> {
.transpose()?
.map(Arc::new);
+ // Compute unified partition type if _partition is projected
+ let unified_partition_type = if field_ids.contains(&RESERVED_FIELD_ID_PARTITION) {
+ let upt = compute_unified_partition_type(
Review Comment:
nit: `upt` seems a bit of unusual in the iceberg-rust repo. how about simply call it `partition_type` or `unified_type`?
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,82 @@ 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 `metadata_columns` map keyed by
+/// field_id. This unified representation (via [`MetadataColumnSource`]) 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>,
+ metadata_columns: HashMap<i32, MetadataColumnSource>,
Review Comment:
hmm, I'd prefer to use constant fields here, which aligns with java's concept. Although, the constant fields are almost always metadata columns. But the identity partition source field is also injected into the constant map.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -376,39 +473,54 @@ 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 metadata_columns.get(field_id) {
+ Some(MetadataColumnSource::Struct(pc)) => {
+ let struct_type = DataType::Struct(pc.fields().clone());
+ let nullable = pc.fields().is_empty();
+ let arrow_field = field_with_id(
+ RESERVED_COL_NAME_PARTITION,
Review Comment:
I think this branch specialized/assumed the `MetadataColumnSource::Struct` must come from the _partition field, which is true for current code but might change when the spec evolves. It would be better to check the field_id is indeed the partition col field first.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,82 @@ 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 `metadata_columns` map keyed by
+/// field_id. This unified representation (via [`MetadataColumnSource`]) 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>,
+ metadata_columns: HashMap<i32, MetadataColumnSource>,
virtual_fields: HashSet<i32>,
}
+/// Unified source for metadata/virtual column constants.
+///
+/// Both scalar metadata columns (like `_file`) and the struct `_partition` column are
+/// per-file constants synthesized during scan. This enum unifies both representations
+/// so the transformer can handle them through a single `metadata_columns` map keyed
+/// by field_id, replacing the previous split between `constant_fields: HashMap<i32, Datum>`
+/// and `partition_column: Option<PartitionColumnConstant>`.
+///
+/// # Design for #2699
+///
+/// When `_pos` is implemented (via arrow-rs RowNumber — a real source column passed through,
+/// not synthesized), it will NOT use this enum. `_pos` is not a constant and belongs in the
+/// virtual_fields / pass-through axis. This enum is exclusively for constant-per-file values.
+///
+/// The `ColumnSource` enum's `Add` and `AddStructConstant` variants map 1:1 to this enum's
+/// variants during transform generation. #2699 may collapse `ColumnSource::Add` and
+/// `ColumnSource::AddStructConstant` into a single `ColumnSource::AddMetadata { source: MetadataColumnSource }`
+/// variant, but that refactor can happen independently.
+#[derive(Debug, Clone, PartialEq)]
+pub enum MetadataColumnSource {
+ /// 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(PartitionColumnConstant),
+}
+
+/// Pre-computed data for the _partition struct constant.
+#[derive(Debug, Clone, PartialEq)]
+pub struct PartitionColumnConstant {
Review Comment:
this could be called as `StructConstant`? It could cover other struct constant if needed, not just the partition column.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5097811343
@CTTY updated (had to force push)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY merged PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3455358721
##########
crates/iceberg/src/metadata_columns.rs:
##########
@@ -379,6 +381,52 @@ pub fn partition_field(partition_fields: Vec<NestedFieldRef>) -> NestedFieldRef
)
}
+/// 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.
+///
+/// For each spec, the partition fields are derived from the transform applied to the source
+/// column. Fields are deduplicated by field_id - each unique field_id appears exactly once
+/// in the result.
+///
+/// # 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 = std::collections::HashSet::new();
+ let mut struct_fields: Vec<NestedFieldRef> = Vec::new();
+
+ for spec in partition_specs {
Review Comment:
Good point!
Updated the implementation (partitioning.rs) with -
- Sort specs by spec_id descending (newer field names take precedence)
- Skips Transform::Void fields (dropped partition columns)
- Deduplicates by field_id
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -116,6 +125,26 @@ impl ManifestEntryContext {
)
.await;
+ // Compute the _partition struct constant if the unified partition type is available
+ let partition_column_constant =
+ if let Some(ref unified_partition_type) = self.unified_partition_type {
+ let partition_spec = self
+ .table_metadata
+ .partition_spec_by_id(self.partition_spec_id);
+ if let Some(spec) = partition_spec {
+ let constant = build_partition_column_constant(
+ unified_partition_type,
+ spec,
+ &self.manifest_entry.data_file.partition,
+ )?;
+ Some(Arc::new(constant))
+ } else {
+ None
+ }
+ } else {
+ None
+ };
Review Comment:
Fair point. Removed `build_partition_column_constant` from scan/context.rs. The scan phase only passes `unified_partition_type` through to the task. The actual struct constant is computed lazily at read time in pipeline.rs.
##########
crates/iceberg/src/metadata_columns.rs:
##########
@@ -379,6 +381,52 @@ pub fn partition_field(partition_fields: Vec<NestedFieldRef>) -> NestedFieldRef
)
}
+/// 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.
+///
+/// For each spec, the partition fields are derived from the transform applied to the source
+/// column. Fields are deduplicated by field_id - each unique field_id appears exactly once
+/// in the result.
+///
+/// # 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:
Created crates/iceberg/src/partitioning.rs, moved the function there, registered as `pub mod partitioning` in lib.rs
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -140,6 +154,13 @@ pub(crate) enum ColumnSource {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
+
+ // A struct column where each child is a constant primitive value.
+ // Used for the _partition metadata column.
+ AddStructConstant {
Review Comment:
Java's approach uses ConstantVectorReader with a StructProjection to map partition data into the unified type. This works because Java's constant map can hold arbitrary values including struct projections. In iceberg-rust, the constant_fields: HashMap<i32, Datum> infrastructure only supports primitive Datum values — Datum has no struct variant. Adding struct support to Datum would be a significant separate effort (touching the spec, serde, and value layers).
The AddStructConstant column source achieves the same functional result through a different mechanism - it builds the struct array directly from the precomputed child values. The per-batch cost is the same (materialize N primitive arrays + wrap in StructArray). If Datum gains struct support in the future, we can consolidated it into the existing constant field infrastructure but for now, the separate variant keeps the change self-contained without requiring changes to the Datum type system.
##########
crates/iceberg/src/spec/schema/_serde.rs:
##########
@@ -100,14 +105,19 @@ impl TryFrom<SchemaV1> for Schema {
}
}
+// Note: alias_to_id is intentionally excluded from serialization.
+// Per Iceberg spec, aliases are not part of the JSON schema representation.
+// They must be reconstructed from external sources after deserialization.
impl From<Schema> for SchemaV2 {
fn from(value: Schema) -> Self {
SchemaV2 {
schema_id: value.schema_id,
identifier_field_ids: if value.identifier_field_ids.is_empty() {
None
} else {
- Some(value.identifier_field_ids.into_iter().collect())
+ let mut ids: Vec<i32> = value.identifier_field_ids.into_iter().collect();
+ ids.sort_unstable();
+ Some(ids)
},
Review Comment:
You're right, this got left over by accident. Removed.
##########
crates/iceberg/src/arrow/value.rs:
##########
@@ -909,6 +909,38 @@ pub(crate) fn create_primitive_array_repeated(
let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
Arc::new(BinaryArray::from_opt_vec(vals))
}
+ (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
+ Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows]))
+ }
+ (DataType::LargeBinary, None) => {
+ let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(LargeBinaryArray::from_opt_vec(vals))
+ }
+ (DataType::FixedSizeBinary(len), Some(PrimitiveLiteral::Binary(value))) => {
+ let repeated: Vec<&[u8]> = vec![value.as_slice(); num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_iter(repeated.into_iter()).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
+ (DataType::FixedSizeBinary(len), None) => {
+ let repeated: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(repeated.into_iter(), *len).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create null FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
Review Comment:
> these seems not related?
>
This is needed, I think. When a table is partitioned by a Binary, UUID, or Time column, `create_primitive_array_repeated` must handle those Arrow types to produce the child arrays of the `_partition` struct.
> Our internal integration also shows iceberg mapping Iceberg's binary to Arrow's LargeBinary though, which should also be updated.
the default `type_to_arrow_type` mapping for Iceberg's `Binary` type should produce `LargeBinary` (matching Java). However, that change affects all `Binary` column reads (not just `_partition`), so it should probably be a follow up.
The `LargeBinary` arm in `create_primitive_array_repeated` ensures we handle it correctly if/when that mapping change lands.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -116,6 +117,21 @@ pub struct FileScanTask {
#[builder(default)]
pub name_mapping: Option<Arc<NameMapping>>,
+ /// Pre-computed constant for the `_partition` metadata struct column.
+ /// Populated by scan planning (or externally by Comet's planner) when
+ /// `RESERVED_FIELD_ID_PARTITION` is in the projected field IDs.
+ ///
+ /// NOTE: This field is not serializable. FileScanTask cannot be
+ /// serialized/deserialized when this is populated. This is consistent with
+ /// partition_spec and name_mapping above, and is acceptable because Comet
+ /// builds tasks in-process (no serialization round-trip).
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[serde(serialize_with = "serialize_not_implemented")]
+ #[serde(deserialize_with = "deserialize_not_implemented")]
+ #[builder(default)]
+ pub partition_column_constant: Option<Arc<PartitionColumnConstant>>,
Review Comment:
`FileScanTask` now carries `unified_partition_type: Option<Arc<StructType>>` instead of `partition_column_constant: Option<Arc<PartitionColumnConstant>>`. The reader computes the value from type + spec + partition data.
Also removed the `table_metadata` field from `ManifestEntryContext` and `ManifestFileContext` (it was only used for the old precomputation path)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4799329577
> We can do the equivalent in Comet - https://github.com/parthchandra/datafusion-comet/blob/iceberg-metadata-columns/native/core/src/execution/planner.rs#L3409-L3440
Are you able to demonstrate this feature working in Comet? Usually when I bring an enhancement to iceberg-rust motivated by Comet usage, I create a draft PR on the Comet repo with tests and the feature built out (using the PR branch here) to demonstrate it works with the Iceberg Java suites.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
CTTY commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3642038275
##########
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:
nit: can we make this error message more explicit? Like
```
format!(
"Struct column constants are only supported for the `_partition` \
metadata column (field id {RESERVED_FIELD_ID_PARTITION}), but one \
was set for field id {field_id}"
)
```
##########
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:
nit: same here, the error message is a bit fuzzy
##########
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:
Can we revisit this? Just read the java impl (`Partitioning::buildPartitionProjectionType`), and my understanding for the void transform handling is:
1. If it's new field, we record its id, type, and name. If we have an existing newer field and the current field is the older field, go to 2
2. We check if the newer field is Void transform AND if the older field is not Void transform, which implies that the field was dropped. In this case, java would preserve the type of the older field. But I don't see this happening in the current function?
##########
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:
We should add unit tests for this function
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
Review Comment:
This feels unneeded since we now have `field_with_id`
##########
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:
This should be pub(crate)
##########
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:
nit: similarly, `build_partition_constant` could be better. We can make the naming consistent
##########
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:
nit: I think `with_partition_constant` is more straightforward
##########
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:
This should be pub(crate)
##########
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:
does this really need to be `pub`? I think a `pub(crate)` is more suitable but not sure if I overlooked anything.
Having 3 loosely coupled arguments for a pub function makes it easier for users mistakes
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3634363914
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -300,6 +303,20 @@ impl<'a> TableScanBuilder<'a> {
.transpose()?
.map(Arc::new);
+ // Compute unified partition type if _partition is projected
+ let unified_partition_type = if field_ids.contains(&RESERVED_FIELD_ID_PARTITION) {
+ let upt = compute_unified_partition_type(
Review Comment:
Changed.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,82 @@ 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 `metadata_columns` map keyed by
+/// field_id. This unified representation (via [`MetadataColumnSource`]) 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>,
+ metadata_columns: HashMap<i32, MetadataColumnSource>,
virtual_fields: HashSet<i32>,
}
+/// Unified source for metadata/virtual column constants.
+///
+/// Both scalar metadata columns (like `_file`) and the struct `_partition` column are
+/// per-file constants synthesized during scan. This enum unifies both representations
+/// so the transformer can handle them through a single `metadata_columns` map keyed
+/// by field_id, replacing the previous split between `constant_fields: HashMap<i32, Datum>`
+/// and `partition_column: Option<PartitionColumnConstant>`.
+///
+/// # Design for #2699
+///
+/// When `_pos` is implemented (via arrow-rs RowNumber — a real source column passed through,
+/// not synthesized), it will NOT use this enum. `_pos` is not a constant and belongs in the
+/// virtual_fields / pass-through axis. This enum is exclusively for constant-per-file values.
+///
+/// The `ColumnSource` enum's `Add` and `AddStructConstant` variants map 1:1 to this enum's
+/// variants during transform generation. #2699 may collapse `ColumnSource::Add` and
+/// `ColumnSource::AddStructConstant` into a single `ColumnSource::AddMetadata { source: MetadataColumnSource }`
+/// variant, but that refactor can happen independently.
+#[derive(Debug, Clone, PartialEq)]
+pub enum MetadataColumnSource {
+ /// 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(PartitionColumnConstant),
+}
+
+/// Pre-computed data for the _partition struct constant.
+#[derive(Debug, Clone, PartialEq)]
+pub struct PartitionColumnConstant {
Review Comment:
changed
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -186,16 +211,82 @@ 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 `metadata_columns` map keyed by
+/// field_id. This unified representation (via [`MetadataColumnSource`]) 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>,
+ metadata_columns: HashMap<i32, MetadataColumnSource>,
Review Comment:
changed. Also renamed the enum from `MetadataColumnSource` to `ColumnConstant` since the values aren't exclusively metadata
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -376,39 +473,54 @@ 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 metadata_columns.get(field_id) {
+ Some(MetadataColumnSource::Struct(pc)) => {
+ let struct_type = DataType::Struct(pc.fields().clone());
+ let nullable = pc.fields().is_empty();
+ let arrow_field = field_with_id(
+ RESERVED_COL_NAME_PARTITION,
Review Comment:
added a check for field id. Also for unexpected struct constants, we now return an error.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3457567533
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -140,6 +154,13 @@ pub(crate) enum ColumnSource {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
+
+ // A struct column where each child is a constant primitive value.
+ // Used for the _partition metadata column.
+ AddStructConstant {
Review Comment:
> In iceberg-rust, the constant_fields: HashMap<i32, Datum> infrastructure only supports primitive Datum values — Datum has no struct variant. Adding struct support to Datum would be a significant separate effort (touching the spec, serde, and value layers).
I see the trade off now. I think it's better to choose your approach now to avoid touching too much internals.
However, I think iceberg-rust already has Struct literal in crates/iceberg/src/spec/values/literal.rs, we can use that to replace the child_values?
And BTW, taking a step forward, the AddStructConstant is essentially the same as
```rust
Add {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
```
which should also be refactor to use `Option<Literal>` in value.
Maybe, we can create a follow-up issue now to refactor these operations. And we can go with your approach for now.
##########
crates/iceberg/src/arrow/value.rs:
##########
@@ -909,6 +909,38 @@ pub(crate) fn create_primitive_array_repeated(
let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
Arc::new(BinaryArray::from_opt_vec(vals))
}
+ (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
+ Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows]))
+ }
+ (DataType::LargeBinary, None) => {
+ let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(LargeBinaryArray::from_opt_vec(vals))
+ }
+ (DataType::FixedSizeBinary(len), Some(PrimitiveLiteral::Binary(value))) => {
+ let repeated: Vec<&[u8]> = vec![value.as_slice(); num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_iter(repeated.into_iter()).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
+ (DataType::FixedSizeBinary(len), None) => {
+ let repeated: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(repeated.into_iter(), *len).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create null FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
Review Comment:
> This is needed, I think. When a table is partitioned by a Binary, UUID, or Time column, create_primitive_array_repeated must handle those Arrow types to produce the child arrays of the _partition struct.
thanks for the explanation.
> the default type_to_arrow_type mapping for Iceberg's Binary type should produce LargeBinary (matching Java)
Actually, java maps Iceberg's Binary to Arrow's Binary at least in the java arrow reader, see https://github.com/apache/iceberg/blob/main/arrow/src/main/java/org/apache/iceberg/arrow/ArrowSchemaUtil.java#L134 and I think parquet-rs reads the binary column in parquet as binary rather than large binary as well.
> However, that change affects all Binary column reads (not just _partition), so it should probably be a follow up.
Yes, it should be addressed in a follow-up.
> The LargeBinary arm in create_primitive_array_repeated ensures we handle it correctly if/when that mapping change lands.
That totally makes sense.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4810605914
Thanks @advancedxy @mbutrovich I'm on the road atm, will address these in a few days.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4937819197
@mbutrovich could you re-review once @hsiang-c 's PRs are merged and I rebase on top of them?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4897645327
@mbutrovich @advancedxy addressed all your review comments.
Sync'd with @hsiang-c and will rebase this on top of #2695 (it will be easier)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3472314907
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -140,6 +154,13 @@ pub(crate) enum ColumnSource {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
+
+ // A struct column where each child is a constant primitive value.
+ // Used for the _partition metadata column.
+ AddStructConstant {
Review Comment:
Thanks for logging the issue.
I was wondering if we can simply use `Struct` type for child _values? It's not required, I'm ok to do that in follow-up.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3502355383
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,80 @@
+// 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 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 = std::collections::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| std::cmp::Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
+ continue;
+ }
+
+ // Skip void transforms (dropped partition columns) and unknown transforms
+ if matches!(field.transform, Transform::Void | Transform::Unknown) {
Review Comment:
fixed
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -359,6 +430,20 @@ impl RecordBatchTransformer {
let fields: Result<Vec<_>> = projected_iceberg_field_ids
.iter()
.map(|field_id| {
+ // Handle _partition struct column
+ if *field_id == RESERVED_FIELD_ID_PARTITION
+ && let Some(pc) = partition_column
+ {
+ let struct_type = DataType::Struct(pc.fields.clone());
+ let nullable = pc.fields.is_empty();
+ let arrow_field = Field::new("_partition", struct_type, nullable)
Review Comment:
fixed
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4848862487
> > We can do the equivalent in Comet - https://github.com/parthchandra/datafusion-comet/blob/iceberg-metadata-columns/native/core/src/execution/planner.rs#L3409-L3440
>
> Are you able to demonstrate this feature working in Comet? Usually when I bring an enhancement to iceberg-rust motivated by Comet usage, I create a draft PR on the Comet repo with tests and the feature built out (using the PR branch here) to demonstrate it works with the Iceberg Java suites.
I've opened a draft PR in Comet: https://github.com/apache/datafusion-comet/pull/4752 based on the same branch I referenced above. All existing tests pass.
The new tests covering this PR are in https://github.com/parthchandra/datafusion-comet/blob/iceberg-metadata-columns/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql
Working on addressing your newer comments (thanks for the detailed review!).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4736759755
@advancedxy fyi
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5124538273
Thank you @CTTY @advancedxy @mbutrovich @hsiang-c !
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3472342846
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,80 @@
+// 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 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 = std::collections::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| std::cmp::Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
+ continue;
+ }
+
+ // Skip void transforms (dropped partition columns) and unknown transforms
+ if matches!(field.transform, Transform::Void | Transform::Unknown) {
Review Comment:
hmm. I don't think this is consistent with iceberg-java.
Unknown transform should be checked rather than ignored.
Considering a new transform introduced in iceberg-spec but not adapted by iceberg-rust yet should be considered as unknown transform.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3472327982
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -359,6 +430,20 @@ impl RecordBatchTransformer {
let fields: Result<Vec<_>> = projected_iceberg_field_ids
.iter()
.map(|field_id| {
+ // Handle _partition struct column
+ if *field_id == RESERVED_FIELD_ID_PARTITION
+ && let Some(pc) = partition_column
+ {
+ let struct_type = DataType::Struct(pc.fields.clone());
+ let nullable = pc.fields.is_empty();
+ let arrow_field = Field::new("_partition", struct_type, nullable)
Review Comment:
nit: should use RESERVED_COL_NAME_PARTITION in metadata_columns directly.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4776525694
@parthchandra I did another round and left some follow-up comments. I think it's in good shape overall.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3676630927
##########
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:
Added back on the scalar constant arm
##########
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:
Done. moved
##########
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:
Done
##########
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:
Fair enough. Replaced with `ok_or_else`.
##########
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:
Thank you for catching this! Added the compatibility check.
##########
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:
Fair point. Changed to `ErrorKind::DataInvalid`
##########
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:
Done. Now errors if the unified type has fields but spec/data is missing.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5106672130
Comet's https://github.com/apache/datafusion-comet/pull/4752 CI is green with this PR.
Comet Iceberg tests for metadata columns are in [CometIcebergNativeSuite.scala](https://github.com/apache/datafusion-comet/pull/4752/changes#diff-7823b8d4877817d000fc9b8407e5a479394b7d0aa2bd3bdfc4befa6cf8530d4d)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5038969297
@advancedxy, @mbutrovich, @CTTY rebased this PR after https://github.com/apache/iceberg-rust/pull/2695 and
https://github.com/apache/iceberg-rust/pull/2746 merged.
The corresponding PR in Comet: https://github.com/apache/datafusion-comet/pull/4752 adds tests as well. (Note the CI failure is an issue in Comet main and not related to this)
PTAL
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4761949653
@parthchandra thanks for pinging me and working on this. I think I'm concerned that the unified partition value is carried in the file scan task, which seems a bit of odd.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3461801364
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,80 @@
+// 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 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 transform fields (dropped partition columns) 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 = std::collections::HashSet::new();
+ let mut struct_fields: Vec<NestedFieldRef> = Vec::new();
Review Comment:
done
##########
crates/iceberg/src/arrow/value.rs:
##########
@@ -909,6 +909,38 @@ pub(crate) fn create_primitive_array_repeated(
let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
Arc::new(BinaryArray::from_opt_vec(vals))
}
+ (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
+ Arc::new(LargeBinaryArray::from_vec(vec![value; num_rows]))
+ }
+ (DataType::LargeBinary, None) => {
+ let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(LargeBinaryArray::from_opt_vec(vals))
+ }
+ (DataType::FixedSizeBinary(len), Some(PrimitiveLiteral::Binary(value))) => {
+ let repeated: Vec<&[u8]> = vec![value.as_slice(); num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_iter(repeated.into_iter()).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
+ (DataType::FixedSizeBinary(len), None) => {
+ let repeated: Vec<Option<&[u8]>> = vec![None; num_rows];
+ Arc::new(FixedSizeBinaryArray::try_from_sparse_iter_with_size(repeated.into_iter(), *len).map_err(|e| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ format!("Failed to create null FixedSizeBinary({len}) array: {e}"),
+ )
+ })?)
+ }
Review Comment:
Logged https://github.com/apache/iceberg-rust/issues/2698 for followup
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -140,6 +154,13 @@ pub(crate) enum ColumnSource {
target_type: DataType,
value: Option<PrimitiveLiteral>,
},
+
+ // A struct column where each child is a constant primitive value.
+ // Used for the _partition metadata column.
+ AddStructConstant {
Review Comment:
Logged https://github.com/apache/iceberg-rust/issues/2699 for follow up
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-4782204085
Comet generates the FileScanTasks from Iceberg Java. Does Iceberg Java provide `unified_partition_type`, otherwise we need to be sure Comet can generate these if you want Comet to work with this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#discussion_r3500804903
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,94 @@
+// 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 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 = std::collections::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| std::cmp::Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
+ continue;
+ }
+
+ // Skip void transforms (dropped partition columns)
+ if matches!(field.transform, Transform::Void) {
+ continue;
+ }
+
+ // Reject unknown transforms — the table uses a spec feature that
+ // this version of iceberg-rust doesn't support. Matching Java's
+ // behavior where getResultType() throws for unknown transforms.
+ 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
+ ),
+ ));
+ }
+
+ seen_field_ids.insert(field.field_id);
+
+ let source_field = schema.field_by_id(field.source_id).ok_or_else(|| {
Review Comment:
Java's `Partitioning.partitionType` only considers `allActiveFieldIds`, which filters out fields whose source column is gone (`schema.findField(field.sourceId()) != null` in `Partitioning.java`). Here we return `ErrorKind::Unexpected` instead. If I'm reading both right, a table that dropped a column it used to partition on (allowed under https://iceberg.apache.org/spec/#partition-evolution) would error the whole scan when `_partition` is projected. Would `continue`-ing on a missing source column match Java's intent better? A test that drops a partition source column would settle it either way.
##########
crates/iceberg/src/arrow/value.rs:
##########
@@ -909,6 +909,38 @@ pub(crate) fn create_primitive_array_repeated(
let vals: Vec<Option<&[u8]>> = vec![None; num_rows];
Arc::new(BinaryArray::from_opt_vec(vals))
}
+ (DataType::LargeBinary, Some(PrimitiveLiteral::Binary(value))) => {
Review Comment:
@advancedxy already questioned whether these arms are related; you explained they're needed for `_partition` struct children and logged the Binary-to-LargeBinary mapping as #2698. This is a separate, reuse-only angle on the same lines, not a relevance question.
I'd push for using arrow-rs here rather than growing our own array-construction match. The hand-rolled arms are duplicate code we own forever: every Arrow type a partition column can take is another arm, and a missing one is a runtime error (the `(dt, _)` catch-all), not a compile error. arrow-rs already maintains correct, exhaustive versions:
- `arrow_array::new_null_array(dt, len)` builds an all-null array for any `DataType`, recursively for `Struct` (`arrow-array/src/array/mod.rs:1020`; struct path `struct_array.rs:192`). This replaces every `(dt, None) => ...` arm and the `(DataType::Struct(fields), None)` arm.
- `PrimitiveArray::from_value(v, n)` (`arrow-array/src/array/primitive_array.rs:808`) for the primitive `Some` arms.
- `GenericByteArray::new_repeated(v, n)` (`arrow-array/src/array/byte_array.rs:198`) for String/Binary/LargeBinary.
Delegating to these shrinks the match substantially and means new partition types are covered by arrow-rs instead of by us adding (and testing) another arm. If there's a reason to keep the explicit arms (e.g. a type arrow-rs handles differently), worth a comment saying so; otherwise I'd lean on the library. A test driving `_partition` through decimal/timestamptz/uuid would also turn any remaining gap into a test failure rather than a read-time error. (Could fold into #2698.)
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,94 @@
+// 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 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 = std::collections::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| std::cmp::Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
+ continue;
+ }
+
+ // Skip void transforms (dropped partition columns)
+ if matches!(field.transform, Transform::Void) {
Review Comment:
Void/unknown handling was already discussed and implemented, so I won't re-litigate it. One residual nuance only if you're interested: because we `continue` on Void before inserting into `seen_field_ids`, if the newest spec marks a field Void and an older spec has it non-Void, the field is still included (good, matches Java's type-upgrade intent), but its name then comes from the older spec, whereas Java prefers the newest spec's name (`Partitioning.java` populates `nameMap` from the first/newest spec that defines the id). Minor and possibly fine as-is; a void-in-newest test would pin whichever behavior you intend.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -240,11 +274,46 @@ impl RecordBatchTransformerBuilder {
Ok(self)
}
+ /// Set the _partition metadata column constant.
+ ///
+ /// This builds the struct constant for the _partition column from the unified partition
+ /// type (across all specs) and the current file's partition data.
+ ///
+ /// # Arguments
+ /// * `unified_partition_type` - The unified partition type across all specs
+ /// * `partition_spec` - The partition spec for this specific file
+ /// * `partition_data` - The partition values for this file
+ #[cfg(test)]
Review Comment:
Since `build_partition_column_constant` is already callable, tests could do `with_partition_column_precomputed(build_partition_column_constant(...)?)`, letting us drop the `#[cfg(test)]` variant and the `_precomputed` suffix. One setter, no test-only API. Reasonable?
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +216,16 @@ pub(crate) struct RecordBatchTransformerBuilder {
snapshot_schema: Arc<IcebergSchema>,
projected_iceberg_field_ids: Vec<i32>,
constant_fields: HashMap<i32, Datum>,
+ partition_column: Option<PartitionColumnConstant>,
Review Comment:
Related to the #2699 follow-up (constant map holds only primitive `Datum`, no struct variant). `_file`/`_spec_id`/identity partitions already synthesize a scalar constant (via `constant_fields`, ending in `ColumnSource::Add`), and `_partition` synthesizes a struct constant (`AddStructConstant`). Rather than adding struct support to `Datum`, one option is to converge the two synthesized paths into a small map:
```
field_id -> MetadataColumnSource {
ScalarConstant(Datum), // _file, _spec_id, identity partitions
StructConstant { fields, child_values }, // _partition
}
```
with `ColumnSource` collapsing toward `FromMetadata { field_id }`. Both arms are test-covered (existing `_file` test plus this PR's `_partition` tests). I'd leave `_pos` (#2746) out: arrow-rs `RowNumber` makes it a real source column that is passed through, not synthesized, so its `virtual_fields` set is a different axis. (@advancedxy floated "use a `Struct` type for the child values" on the related thread; the enum is a way to do that without widening `Datum`.) Not a blocker for this PR; feel free to fold into #2699 if that's a better home.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -359,6 +432,21 @@ impl RecordBatchTransformer {
let fields: Result<Vec<_>> = projected_iceberg_field_ids
.iter()
.map(|field_id| {
+ // Handle _partition struct column
+ if *field_id == RESERVED_FIELD_ID_PARTITION
+ && let Some(pc) = partition_column
+ {
+ let struct_type = DataType::Struct(pc.fields.clone());
+ let nullable = pc.fields.is_empty();
+ let arrow_field =
+ Field::new(RESERVED_COL_NAME_PARTITION, struct_type, nullable)
Review Comment:
This exact block recurs several times across these PRs. A small `fn field_with_id(name, dt, nullable, field_id) -> Arc<Field>` helper would remove the repetition and keep the field-id metadata format in one place.
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -138,6 +142,7 @@ impl ManifestEntryContext {
// TODO: Pass actual PartitionSpec through context chain for native flow
.with_partition_spec(None)
Review Comment:
`build_partition_column_constant` needs `task.partition_spec` to be `Some` for partitioned tables, but this PR keeps `partition_spec(None)`; #2695 is the one that wires the real spec through `context.rs`. Is the plan to rebase on #2695? If so, an end-to-end test asserting non-null partition values would confirm the wiring once rebased.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -300,6 +303,20 @@ impl<'a> TableScanBuilder<'a> {
.transpose()?
.map(Arc::new);
+ // Compute unified partition type if _partition is projected
Review Comment:
Gating on "`_partition` projected" is a nice touch. Java builds the type from `table.specs().values()` (all specs). Could we confirm `partition_specs_iter()` enumerates every historical spec, not just current, and ideally pin it with a multi-spec test? That is what makes evolution visible in `_partition`.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -193,6 +216,16 @@ pub(crate) struct RecordBatchTransformerBuilder {
snapshot_schema: Arc<IcebergSchema>,
projected_iceberg_field_ids: Vec<i32>,
constant_fields: HashMap<i32, Datum>,
+ partition_column: Option<PartitionColumnConstant>,
+}
+
+/// Pre-computed data for the _partition struct constant.
+#[derive(Debug, Clone, PartialEq)]
+pub struct PartitionColumnConstant {
+ /// Arrow struct fields (names, types, nullability) for the _partition column.
+ pub fields: Fields,
Review Comment:
These two are assumed equal-length (the `zip` in `create_struct_column` silently truncates otherwise). A `new(fields, child_values)` constructor plus accessors would let us assert that invariant in one place and keeps with the crate's "no public fields" convention. Minor, but it closes a small footgun.
##########
crates/iceberg/src/arrow/record_batch_transformer.rs:
##########
@@ -656,6 +761,84 @@ impl RecordBatchTransformer {
create_primitive_array_repeated(target_type, prim_lit, num_rows)
}
}
+
+ fn create_struct_column(
Review Comment:
`create_struct_column` is hand-rolled struct assembly that arrow-rs already provides: the empty/all-null path is `StructArray::new_null(fields, len)` / `new_null_array` (`arrow-array/src/array/struct_array.rs:192`), and the populated path is `StructArray::try_new(fields, arrays, nulls)`, which validates `fields.len() == arrays.len()` and child-length agreement (`struct_array.rs:106`). I'd reuse those rather than maintain our own constructor: `try_new` also enforces the `fields`/`child_values` length invariant that the current `zip` silently truncates. Same principle as the `value.rs` note: prefer the library's validated constructors over duplicating them here.
##########
crates/iceberg/src/arrow/reader/pipeline.rs:
##########
@@ -255,6 +260,30 @@ impl FileScanTaskReader {
record_batch_transformer_builder.with_partition(partition_spec, partition_data)?;
}
+ // 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
+ {
+ if unified_type.fields().is_empty() {
Review Comment:
The "unpartitioned becomes empty struct becomes null" rule (which matches Java `PartitionUtil.constantsMap`, where an empty `partitionType` stores `null`) currently lives in the orchestration layer. If `build_partition_column_constant` handled the empty case internally and returned the empty constant, `pipeline.rs` could always call one entry point and drop the branch, and a direct caller couldn't forget the null case. Thoughts?
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,94 @@
+// 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 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 = std::collections::HashSet::new();
Review Comment:
Small consistency thing: `use std::collections::HashSet;` / `use std::cmp::Reverse;` up top matches the surrounding style.
##########
crates/iceberg/src/partitioning.rs:
##########
@@ -0,0 +1,94 @@
+// 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 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 = std::collections::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| std::cmp::Reverse(s.spec_id()));
+
+ for spec in specs {
+ for field in spec.fields() {
+ if seen_field_ids.contains(&field.field_id) {
+ continue;
+ }
+
+ // Skip void transforms (dropped partition columns)
+ if matches!(field.transform, Transform::Void) {
+ continue;
+ }
+
+ // Reject unknown transforms — the table uses a spec feature that
+ // this version of iceberg-rust doesn't support. Matching Java's
+ // behavior where getResultType() throws for unknown transforms.
+ 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
+ ),
+ ));
+ }
+
+ seen_field_ids.insert(field.field_id);
+
+ let source_field = schema.field_by_id(field.source_id).ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ format!(
+ "No column with source column id {} in schema for partition field {}",
+ field.source_id, field.name
+ ),
+ )
+ })?;
+
+ let res_type = field.transform.result_type(&source_field.field_type)?;
+ let nested = NestedField::optional(field.field_id, &field.name, res_type).into();
+ struct_fields.push(nested);
+ }
+ }
+
+ Ok(StructType::new(struct_fields))
Review Comment:
`buildPartitionProjectionType` sorts its output fields by id ascending (`fieldMap.keySet().stream().sorted(Comparator.naturalOrder())` in `Partitioning.java`). Here we emit them in spec-descending, first-seen order. In the common case these coincide, but under partition evolution I think they can differ, and since the `_partition` struct schema is consumed by engines (this is the Comet path), matching Java's ordering seems safer. Worth a sort plus an evolution test that would otherwise reorder? Curious whether you already considered this and decided it's fine.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -116,6 +116,17 @@ pub struct FileScanTask {
#[builder(default)]
pub name_mapping: Option<Arc<NameMapping>>,
+ /// The unified partition type across all specs in the table.
+ /// When `RESERVED_FIELD_ID_PARTITION` is in the projected field IDs, the reader
+ /// uses this type along with the task's partition_spec and partition data to
+ /// materialize the `_partition` struct column at read time.
+ #[serde(default)]
+ #[serde(skip_serializing_if = "Option::is_none")]
+ #[serde(serialize_with = "serialize_not_implemented")]
Review Comment:
This connects to your own thread on the PR (Comet doesn't get `unified_partition_type` from Iceberg Java, and parthchandra's plan is for Comet to compute the equivalent, per the linked planner.rs). Independent of that: the serde attrs here match the existing `partition` / `partition_spec` / `name_mapping` fields, so it's consistent, not a regression. The implication worth one line in the PR text: this field (like the others) does not round-trip through serde, so the `_partition` machinery works in the native scan flow only. Since Comet builds tasks natively that's fine, just easy for a future reader to assume otherwise.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "CTTY (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "advancedxy (via GitHub)" <gi...@apache.org>.
advancedxy commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5034055350
@parthchandra since other prs are merged, would you like to rebase the PR on the master and let's continue.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5097472022
@CTTY [6bf928a](https://github.com/apache/iceberg-rust/pull/2668/commits/6bf928af1585c0e35d65b8c385dec48f09f59f6a) has the changes but they don't seem to be showing up. Let me rebase and push again.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: add support for _partition metadata column [iceberg-rust]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on PR #2668:
URL: https://github.com/apache/iceberg-rust/pull/2668#issuecomment-5060959699
Thanks @advancedxy. @CTTY could we get a committer review/approval?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org