You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by gi...@apache.org on 2026/10/22 13:50:15 UTC
(iceberg-rust) branch gh-readonly-queue/main/pr-3431-86e837f90d16074d67218db05ffc2629443dafd5 created (now 1c05452b5)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-3431-86e837f90d16074d67218db05ffc2629443dafd5
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
at 1c05452b5 fix(spec): accept partition specs produced by spec evolution (#3431)
This branch includes the following new commits:
new 1c05452b5 fix(spec): accept partition specs produced by spec evolution (#3431)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iceberg-rust) 01/01: fix(spec): accept partition specs produced by spec evolution (#3431)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3431-86e837f90d16074d67218db05ffc2629443dafd5
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 1c05452b53f65d4d37f6399c550ca0d2413ded0c
Author: Dhruv Arya <dh...@databricks.com>
AuthorDate: Thu Oct 22 13:49:50 2026 +0000
fix(spec): accept partition specs produced by spec evolution (#3431)
* fix(spec): accept partition specs produced by spec evolution
Partition spec evolution produces specs that the spec builders rejected:
- A source column can have more than one partition field. Adding hour(ts) to a
spec that has day(ts) is allowed, and format version 1 tables keep each
removed field as a `void` placeholder. Binding an unbound spec
(`UnboundPartitionSpec::bind`, `PartitionSpecBuilder::new_from_unbound`, and
so `TableMetadataBuilder::add_partition_spec`) and loading the legacy v1
`partition-spec` field ran the authoring-time redundancy check and rejected
these specs. Both now skip it, as Java does.
`PartitionSpecBuilder::add_unbound_field` and
`UnboundPartitionSpecBuilder::add_partition_field` still reject redundant
fields.
- Removing identity(category) from a format version 1 spec leaves a void field
named `category`. Both the spec builder and the table metadata builder
allowed only identity fields to share the name of their source column. Void
fields may now do the same, matching Java's `alwaysNull`.
Co-authored-by: Isaac <no...@databricks.com>
* fix(spec): address review feedback on evolved spec binding
Share the identity-or-void name rule through `Transform::may_share_name_with_source`
and name void in its error. Check names before redundancy in `add_unbound_field`,
document the authoring-vs-bind split, and pin it with tests, including a v2
day+hour default spec that survives later commits.
Co-authored-by: Isaac <no...@databricks.com>
---------
Co-authored-by: Dhruv Arya <ar...@gmail.com>
Co-authored-by: Isaac <no...@databricks.com>
Co-authored-by: Kevin Liu <ke...@users.noreply.github.com>
---
crates/iceberg/src/spec/partition.rs | 168 +++++++++++++++++++++-
crates/iceberg/src/spec/table_metadata.rs | 41 +++++-
crates/iceberg/src/spec/table_metadata_builder.rs | 102 +++++++++++--
crates/iceberg/src/spec/transform.rs | 8 ++
4 files changed, 300 insertions(+), 19 deletions(-)
diff --git a/crates/iceberg/src/spec/partition.rs b/crates/iceberg/src/spec/partition.rs
index fdb5a4a7b..3e639de00 100644
--- a/crates/iceberg/src/spec/partition.rs
+++ b/crates/iceberg/src/spec/partition.rs
@@ -437,6 +437,9 @@ impl UnboundPartitionSpec {
}
/// Bind this unbound partition spec to a schema.
+ ///
+ /// Unlike [`PartitionSpecBuilder::add_unbound_field`], this accepts redundant fields, such
+ /// as two time granularities of one column, because spec evolution produces them.
pub fn bind(self, schema: impl Into<SchemaRef>) -> Result<PartitionSpec> {
PartitionSpecBuilder::new_from_unbound(self, schema)?.build()
}
@@ -578,7 +581,7 @@ impl PartitionSpecBuilder {
Self::new(schema).with_spec_id(unbound.spec_id.unwrap_or(DEFAULT_PARTITION_SPEC_ID));
for field in unbound.fields {
- builder = builder.add_unbound_field(field)?;
+ builder = builder.add_field_allowing_redundancy(field)?;
}
Ok(builder)
}
@@ -630,9 +633,20 @@ impl PartitionSpecBuilder {
///
/// If partition field id is set, it is used as the field id.
/// Otherwise, a new `field_id` is assigned.
- pub fn add_unbound_field(mut self, field: UnboundPartitionField) -> Result<Self> {
+ ///
+ /// Rejects a field that is redundant with an existing one, such as a second time
+ /// granularity of the same column. [`UnboundPartitionSpec::bind`] accepts such fields.
+ pub fn add_unbound_field(self, field: UnboundPartitionField) -> Result<Self> {
self.check_name_set_and_unique(&field.name)?;
self.check_for_redundant_partitions(&field.source_ids, &field.transform)?;
+ self.add_field_allowing_redundancy(field)
+ }
+
+ /// Same as [`Self::add_unbound_field`] without the redundancy check, as Java's
+ /// `PartitionSpec.Builder.add` does. Used to bind persisted specs, where evolution can leave
+ /// several time granularities or void fields for one column.
+ fn add_field_allowing_redundancy(mut self, field: UnboundPartitionField) -> Result<Self> {
+ self.check_name_set_and_unique(&field.name)?;
Self::check_name_does_not_collide_with_schema(&field, &self.schema)?;
Self::check_transform_compatibility(&field, &self.schema)?;
if let Some(partition_field_id) = field.field_id {
@@ -709,19 +723,19 @@ impl PartitionSpecBuilder {
/// Ensure that the partition name is unique among columns in the schema.
/// Duplicate names are allowed if:
/// 1. The column is sourced from the column with the same name.
- /// 2. AND the transformation is identity
+ /// 2. AND [`Transform::may_share_name_with_source`] holds for the transform.
fn check_name_does_not_collide_with_schema(
field: &UnboundPartitionField,
schema: &Schema,
) -> Result<()> {
match schema.field_by_name(field.name.as_str()) {
Some(schema_collision) => {
- if field.transform == Transform::Identity {
+ if field.transform.may_share_name_with_source() {
if schema_collision.id == field.source_id()? {
Ok(())
} else {
Err(invalid_data!(
- "Cannot create identity partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`",
+ "Cannot create identity or void partition sourced from different field in schema. Field name '{}' has id `{}` in schema but partition source id is `{}`",
field.name,
schema_collision.id,
field.source_id()?
@@ -729,7 +743,7 @@ impl PartitionSpecBuilder {
}
} else {
Err(invalid_data!(
- "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity transform.",
+ "Cannot create partition with name: '{}' that conflicts with schema field and is not an identity or void transform.",
field.name
))
}
@@ -1564,6 +1578,148 @@ mod tests {
assert!(err.message().contains("redundant partition"));
}
+ fn ts_schema() -> Schema {
+ Schema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(),
+ NestedField::required(2, "id", Type::Primitive(PrimitiveType::Long)).into(),
+ ])
+ .build()
+ .unwrap()
+ }
+
+ fn ts_field(field_id: i32, name: &str, transform: Transform) -> UnboundPartitionField {
+ UnboundPartitionField {
+ source_ids: vec![1],
+ field_id: Some(field_id),
+ name: name.to_string(),
+ transform,
+ }
+ }
+
+ #[test]
+ fn test_add_unbound_field_rejects_second_time_granularity() {
+ let err = PartitionSpec::builder(ts_schema())
+ .add_unbound_field(ts_field(1000, "ts_day", Transform::Day))
+ .unwrap()
+ .add_unbound_field(ts_field(1001, "ts_hour", Transform::Hour))
+ .unwrap_err();
+ assert!(err.message().contains("redundant partition"));
+ }
+
+ #[test]
+ fn test_bind_accepts_evolved_spec_fields() {
+ let fields = vec![
+ ts_field(1000, "ts_day", Transform::Day),
+ ts_field(1001, "ts_hour", Transform::Hour),
+ ts_field(1002, "ts", Transform::Void),
+ ts_field(1003, "ts_null", Transform::Void),
+ ];
+ let unbound = UnboundPartitionSpec {
+ spec_id: Some(1),
+ fields: fields.clone(),
+ };
+
+ let spec = unbound.bind(ts_schema()).unwrap();
+
+ let expected = PartitionSpec {
+ spec_id: 1,
+ fields: fields
+ .into_iter()
+ .map(|f| PartitionField {
+ source_id: 1,
+ field_id: f.field_id.unwrap(),
+ name: f.name,
+ transform: f.transform,
+ })
+ .collect(),
+ };
+ assert_eq!(spec, expected);
+ }
+
+ #[test]
+ fn test_bind_rejects_void_field_named_after_other_column() {
+ let unbound = UnboundPartitionSpec {
+ spec_id: Some(1),
+ fields: vec![ts_field(1000, "id", Transform::Void)],
+ };
+
+ let err = unbound.bind(ts_schema()).unwrap_err();
+
+ assert_eq!(err.kind(), ErrorKind::DataInvalid);
+ assert!(err.message().contains("Field name 'id'"));
+ }
+
+ #[test]
+ fn test_bind_accepts_identical_fields_with_different_names() {
+ let bucket_field = |field_id: i32, name: &str| UnboundPartitionField {
+ source_ids: vec![2],
+ field_id: Some(field_id),
+ name: name.to_string(),
+ transform: Transform::Bucket(16),
+ };
+ let unbound = UnboundPartitionSpec {
+ spec_id: Some(1),
+ fields: vec![
+ bucket_field(1000, "id_bucket"),
+ bucket_field(1001, "id_bucket_2"),
+ ],
+ };
+
+ let spec = unbound.bind(ts_schema()).unwrap();
+
+ let expected = PartitionSpec {
+ spec_id: 1,
+ fields: vec![
+ PartitionField {
+ source_id: 2,
+ field_id: 1000,
+ name: "id_bucket".to_string(),
+ transform: Transform::Bucket(16),
+ },
+ PartitionField {
+ source_id: 2,
+ field_id: 1001,
+ name: "id_bucket_2".to_string(),
+ transform: Transform::Bucket(16),
+ },
+ ],
+ };
+ assert_eq!(spec, expected);
+ }
+
+ #[test]
+ fn test_bind_rejects_repeated_partition_name() {
+ let unbound = UnboundPartitionSpec {
+ spec_id: Some(1),
+ fields: vec![
+ ts_field(1000, "ts_part", Transform::Day),
+ ts_field(1001, "ts_part", Transform::Hour),
+ ],
+ };
+
+ let err = unbound.bind(ts_schema()).unwrap_err();
+
+ assert!(
+ err.message()
+ .contains("Cannot use partition name more than once")
+ );
+ }
+
+ #[test]
+ fn test_add_unbound_field_checks_name_before_redundancy() {
+ let err = PartitionSpec::builder(ts_schema())
+ .add_unbound_field(ts_field(1000, "ts_part", Transform::Day))
+ .unwrap()
+ .add_unbound_field(ts_field(1001, "ts_part", Transform::Hour))
+ .unwrap_err();
+
+ assert!(
+ err.message()
+ .contains("Cannot use partition name more than once")
+ );
+ }
+
#[test]
fn test_builder_incompatible_transforms_disallowed() {
let schema = Schema::builder()
diff --git a/crates/iceberg/src/spec/table_metadata.rs b/crates/iceberg/src/spec/table_metadata.rs
index 83468602c..99df566f6 100644
--- a/crates/iceberg/src/spec/table_metadata.rs
+++ b/crates/iceberg/src/spec/table_metadata.rs
@@ -785,7 +785,7 @@ pub(super) mod _serde {
use crate::spec::{
EncryptedKey, INITIAL_ROW_ID, PartitionField, PartitionSpec, PartitionSpecRef,
PartitionStatisticsFile, Schema, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention,
- SortOrder, StatisticsFile,
+ SortOrder, StatisticsFile, UnboundPartitionSpec,
};
use crate::{Error, ErrorKind};
@@ -1220,10 +1220,14 @@ pub(super) mod _serde {
.collect::<HashMap<_, _>>()
} else if let Some(partition_spec) = value.partition_spec {
// Option 2: Fall back to 'partition_spec'
- let spec = PartitionSpec::builder(current_schema.clone())
- .with_spec_id(DEFAULT_PARTITION_SPEC_ID)
- .add_unbound_fields(partition_spec.into_iter().map(|f| f.into_unbound()))?
- .build()?;
+ let spec = UnboundPartitionSpec {
+ spec_id: Some(DEFAULT_PARTITION_SPEC_ID),
+ fields: partition_spec
+ .into_iter()
+ .map(|f| f.into_unbound())
+ .collect(),
+ }
+ .bind(current_schema.clone())?;
HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))])
} else {
@@ -3317,6 +3321,33 @@ mod tests {
check_table_metadata_serde(&metadata, expected);
}
+ #[test]
+ fn test_table_metadata_v1_legacy_partition_spec_with_void_fields() {
+ let mut metadata: serde_json::Value = serde_json::from_str(
+ &fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap(),
+ )
+ .unwrap();
+ // What a format version 1 spec keeps after identity(x) was removed twice.
+ metadata["partition-spec"] = serde_json::json!([
+ {"name": "x", "transform": "void", "source-id": 1, "field-id": 1000},
+ {"name": "x_null", "transform": "void", "source-id": 1, "field-id": 1001},
+ ]);
+
+ let metadata: TableMetadata = serde_json::from_value(metadata).unwrap();
+
+ let fields = metadata.default_partition_spec().fields();
+ assert_eq!(
+ fields
+ .iter()
+ .map(|f| (f.field_id, f.name.as_str(), f.transform))
+ .collect::<Vec<_>>(),
+ vec![
+ (1000, "x", Transform::Void),
+ (1001, "x_null", Transform::Void)
+ ]
+ );
+ }
+
#[test]
fn test_empty_snapshot_id_is_normalized_to_none() {
let metadata =
diff --git a/crates/iceberg/src/spec/table_metadata_builder.rs b/crates/iceberg/src/spec/table_metadata_builder.rs
index 44deb3bfe..a8ef84a18 100644
--- a/crates/iceberg/src/spec/table_metadata_builder.rs
+++ b/crates/iceberg/src/spec/table_metadata_builder.rs
@@ -726,12 +726,13 @@ impl TableMetadataBuilder {
/// Validate partition field names against schema field names across all historical schemas.
///
/// Due to Iceberg's multi-version property, partition fields can share names with schema fields
- /// if they meet specific requirements (identity transform + matching source field ID).
+ /// if they meet specific requirements (identity or void transform + matching source field ID).
/// This validation enforces those rules across all historical schema versions.
///
/// # Errors
- /// - Partition field name conflicts with schema field name but doesn't use identity transform.
- /// - Partition field uses identity transform but references wrong source field ID.
+ /// - Partition field name conflicts with schema field name but doesn't use identity or void
+ /// transform.
+ /// - Partition field uses identity or void transform but references wrong source field ID.
fn validate_partition_field_names(&self, unbound_spec: &UnboundPartitionSpec) -> Result<()> {
if self.metadata.schemas.is_empty() {
return Ok(());
@@ -750,19 +751,18 @@ impl TableMetadataBuilder {
// If name exists in schemas, validate against current schema rules
if let Some(schema_field) = current_schema.field_by_name(partition_field.name()) {
- let is_identity_transform = partition_field.transform() == Transform::Identity;
let has_matching_source_id = schema_field.id == partition_field.source_id()?;
- if !is_identity_transform {
+ if !partition_field.transform().may_share_name_with_source() {
return Err(invalid_data!(
- "Cannot create partition with name '{}' that conflicts with schema field and is not an identity transform.",
+ "Cannot create partition with name '{}' that conflicts with schema field and is not an identity or void transform.",
partition_field.name()
));
}
if !has_matching_source_id {
return Err(invalid_data!(
- "Cannot create identity partition sourced from different field in schema. \
+ "Cannot create identity or void partition sourced from different field in schema. \
Field name '{}' has id `{}` in schema but partition source id is `{}`",
partition_field.name(),
schema_field.id,
@@ -2471,6 +2471,92 @@ mod tests {
));
}
+ #[test]
+ fn test_add_partition_spec_accepts_void_fields_of_removed_fields() {
+ let builder = builder_without_changes(FormatVersion::V1);
+ let void_field = |field_id: i32, name: &str| {
+ UnboundPartitionField::builder()
+ .source_ids(vec![2])
+ .field_id(field_id)
+ .name(name)
+ .transform(Transform::Void)
+ .build()
+ .unwrap()
+ };
+ let added_spec = UnboundPartitionSpec {
+ spec_id: None,
+ fields: vec![void_field(1000, "y"), void_field(1001, "y_null")],
+ };
+
+ let metadata = builder
+ .add_default_partition_spec(added_spec.clone())
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+
+ let expected = added_spec.with_spec_id(1).bind(schema()).unwrap();
+ assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
+ assert_eq!(metadata.last_partition_id, 1001);
+ }
+
+ #[test]
+ fn test_add_partition_spec_accepts_two_time_granularities_of_one_column() {
+ let schema = Schema::builder()
+ .with_fields(vec![
+ NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(),
+ ])
+ .build()
+ .unwrap();
+ let time_field = |field_id: i32, name: &str, transform: Transform| {
+ UnboundPartitionField::builder()
+ .source_ids(vec![1])
+ .field_id(field_id)
+ .name(name)
+ .transform(transform)
+ .build()
+ .unwrap()
+ };
+ let added_spec = UnboundPartitionSpec {
+ spec_id: None,
+ fields: vec![
+ time_field(1000, "ts_day", Transform::Day),
+ time_field(1001, "ts_hour", Transform::Hour),
+ ],
+ };
+ let metadata = TableMetadataBuilder::new(
+ schema.clone(),
+ PartitionSpec::unpartition_spec(),
+ SortOrder::unsorted_order(),
+ TEST_LOCATION.to_string(),
+ FormatVersion::V2,
+ HashMap::new(),
+ )
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+
+ let metadata = metadata
+ .into_builder(None)
+ .add_default_partition_spec(added_spec.clone())
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+ // Every later commit re-validates the default spec.
+ let metadata = metadata
+ .into_builder(None)
+ .set_properties(HashMap::from([("key".to_string(), "value".to_string())]))
+ .unwrap()
+ .build()
+ .unwrap()
+ .metadata;
+
+ let expected = added_spec.with_spec_id(1).bind(schema).unwrap();
+ assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
+ }
+
#[test]
fn test_expire_metadata_log() {
let builder = builder_without_changes(FormatVersion::V2);
@@ -2957,7 +3043,7 @@ mod tests {
assert!(error_message.contains(
"Cannot create partition with name 'existing_field' that conflicts with schema field"
));
- assert!(error_message.contains("and is not an identity transform"));
+ assert!(error_message.contains("and is not an identity or void transform"));
}
#[test]
diff --git a/crates/iceberg/src/spec/transform.rs b/crates/iceberg/src/spec/transform.rs
index 0d2b61e73..f52202b78 100644
--- a/crates/iceberg/src/spec/transform.rs
+++ b/crates/iceberg/src/spec/transform.rs
@@ -404,6 +404,14 @@ impl Transform {
)
}
+ /// Whether a partition field with this transform may be named after its source column.
+ ///
+ /// Void is allowed because removing an identity field from a format version 1 spec leaves
+ /// a void field that keeps the identity field's name.
+ pub(crate) fn may_share_name_with_source(&self) -> bool {
+ matches!(self, Transform::Identity | Transform::Void)
+ }
+
/// Return the unique transform name to check if similar transforms for the same source field
/// are added multiple times in partition spec builder.
pub fn dedup_name(&self) -> String {