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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
