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: [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]
