From c11ef4e25fe5f07722dca43b6b46c269bb15909e Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Tue, 10 Feb 2026 14:32:25 +0200 Subject: [PATCH 01/10] add update_schema --- crates/iceberg/src/transaction/mod.rs | 7 + .../iceberg/src/transaction/update_schema.rs | 1193 +++++++++++++++++ .../integration_tests/tests/update_schema.rs | 409 ++++++ 3 files changed, 1609 insertions(+) create mode 100644 crates/iceberg/src/transaction/update_schema.rs create mode 100644 crates/integration_tests/tests/update_schema.rs diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index cdeeb2abc9..9bf569ac95 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -59,6 +59,7 @@ mod snapshot; mod sort_order; mod update_location; mod update_properties; +mod update_schema; mod update_statistics; mod upgrade_format_version; @@ -76,6 +77,7 @@ use crate::transaction::rollback_to_snapshot::RollbackToSnapshotAction; use crate::transaction::sort_order::ReplaceSortOrderAction; use crate::transaction::update_location::UpdateLocationAction; use crate::transaction::update_properties::UpdatePropertiesAction; +use crate::transaction::update_schema::UpdateSchemaAction; use crate::transaction::update_statistics::UpdateStatisticsAction; use crate::transaction::upgrade_format_version::UpgradeFormatVersionAction; use crate::{Catalog, TableCommit, TableRequirement, TableUpdate}; @@ -138,6 +140,11 @@ impl Transaction { UpdatePropertiesAction::new() } + /// Creates an update schema action. + pub fn update_schema(&self) -> UpdateSchemaAction { + UpdateSchemaAction::new() + } + /// Creates a fast append action. pub fn fast_append(&self) -> FastAppendAction { FastAppendAction::new() diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs new file mode 100644 index 0000000000..3caa275be1 --- /dev/null +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -0,0 +1,1193 @@ +// 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. + +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; + +use async_trait::async_trait; + +use crate::spec::{ + ListType, Literal, MapType, NestedField, NestedFieldRef, Schema, StructType, Type, +}; +use crate::table::Table; +use crate::transaction::action::{ActionCommit, TransactionAction}; +use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate}; + +/// Sentinel parent ID representing the table root (top-level columns). +const TABLE_ROOT_ID: i32 = -1; + +/// A pending column addition, recording the parent path and the field to add. +struct PendingAdd { + /// `None` means a root-level addition; `Some("person")` or `Some("person.address")` + /// identifies the nested struct to add the column to. + parent: Option, + /// The field to add. Uses placeholder ID `0` which is auto-assigned at commit time. + field: NestedFieldRef, +} + +/// Schema evolution API modeled after the Java `SchemaUpdate` implementation. +/// +/// This action accumulates schema modifications (column additions and deletions) +/// via builder methods. At commit time, it validates all operations against the +/// current table schema, auto-assigns field IDs from `table.metadata().last_column_id()`, +/// builds a new schema, and emits `AddSchema` + `SetCurrentSchema` updates with a +/// `CurrentSchemaIdMatch` requirement. +/// +/// # Example +/// +/// ```ignore +/// let tx = Transaction::new(&table); +/// let action = tx.update_schema() +/// .add_column("new_col", Type::Primitive(PrimitiveType::Int)) +/// .add_column_to("person", "email", Type::Primitive(PrimitiveType::String)) +/// .delete_column("old_col"); +/// let tx = action.apply(tx).unwrap(); +/// let table = tx.commit(&catalog).await.unwrap(); +/// ``` +pub struct UpdateSchemaAction { + additions: Vec, + deletes: Vec, + auto_assign_ids: bool, +} + +impl UpdateSchemaAction { + /// Creates a new empty `UpdateSchemaAction`. + pub(crate) fn new() -> Self { + Self { + additions: Vec::new(), + deletes: Vec::new(), + auto_assign_ids: true, + } + } + + // --- Root-level additions --- + + /// Add a `NestedFieldRef` column to the table root. + pub fn add_field(self, field: NestedFieldRef) -> Self { + self.add_field_internal(None, field) + } + + /// Add an optional column to the table root. + /// + /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. + pub fn add_column(self, name: impl ToString, field_type: Type) -> Self { + self.add_field(Arc::new(NestedField::optional(0, name, field_type))) + } + + /// Add an optional column with a doc string to the table root. + /// + /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. + pub fn add_column_with_doc( + self, + name: impl ToString, + field_type: Type, + doc: impl ToString, + ) -> Self { + self.add_field(Arc::new( + NestedField::optional(0, name, field_type).with_doc(doc), + )) + } + + /// Add a required column to the table root. + /// + /// An `initial_default` value is required per the Iceberg spec: it is used to populate + /// this field for all records that were written before the field was added. + /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. + pub fn add_required_column( + self, + name: impl ToString, + field_type: Type, + initial_default: Literal, + ) -> Self { + self.add_field(Arc::new( + NestedField::required(0, name, field_type).with_initial_default(initial_default), + )) + } + + /// Add a required column with a doc string to the table root. + /// + /// An `initial_default` value is required per the Iceberg spec: it is used to populate + /// this field for all records that were written before the field was added. + /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. + pub fn add_required_column_with_doc( + self, + name: impl ToString, + field_type: Type, + initial_default: Literal, + doc: impl ToString, + ) -> Self { + self.add_field(Arc::new( + NestedField::required(0, name, field_type) + .with_initial_default(initial_default) + .with_doc(doc), + )) + } + + // --- Nested additions --- + + /// Add a `NestedFieldRef` column under a parent struct identified by name. + /// + /// If the parent is a map, the column is added to the map value's struct. + /// If the parent is a list, the column is added to the list element's struct. + pub fn add_field_to(self, parent: impl ToString, field: NestedFieldRef) -> Self { + self.add_field_internal(Some(parent.to_string()), field) + } + + /// Add an optional column under a parent struct identified by name. + /// + /// The `parent` can be a dotted path (e.g. `"person"` or `"person.address"`). + /// If the parent is a map, the column is added to the map value's struct. + /// If the parent is a list, the column is added to the list element's struct. + /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. + pub fn add_column_to( + self, + parent: impl ToString, + name: impl ToString, + field_type: Type, + ) -> Self { + self.add_field_to(parent, Arc::new(NestedField::optional(0, name, field_type))) + } + + /// Add an optional column with a doc string under a parent struct. + /// + /// See [`add_column_to`](Self::add_column_to) for parent path details. + pub fn add_column_to_with_doc( + self, + parent: impl ToString, + name: impl ToString, + field_type: Type, + doc: impl ToString, + ) -> Self { + self.add_field_to( + parent, + Arc::new(NestedField::optional(0, name, field_type).with_doc(doc)), + ) + } + + /// Add a required column under a parent struct. + /// + /// See [`add_column_to`](Self::add_column_to) for parent path details. + /// An `initial_default` value is required per the Iceberg spec. + pub fn add_required_column_to( + self, + parent: impl ToString, + name: impl ToString, + field_type: Type, + initial_default: Literal, + ) -> Self { + self.add_field_to( + parent, + Arc::new( + NestedField::required(0, name, field_type).with_initial_default(initial_default), + ), + ) + } + + /// Add a required column with a doc string under a parent struct. + /// + /// See [`add_column_to`](Self::add_column_to) for parent path details. + /// An `initial_default` value is required per the Iceberg spec. + pub fn add_required_column_to_with_doc( + self, + parent: impl ToString, + name: impl ToString, + field_type: Type, + initial_default: Literal, + doc: impl ToString, + ) -> Self { + self.add_field_to( + parent, + Arc::new( + NestedField::required(0, name, field_type) + .with_initial_default(initial_default) + .with_doc(doc), + ), + ) + } + + // --- Other builder methods --- + + /// Record a column deletion by name. + /// + /// At commit time, the column must exist in the current schema. + pub fn delete_column(mut self, name: impl ToString) -> Self { + self.deletes.push(name.to_string()); + self + } + + /// Disable automatic field ID assignment. When disabled, the placeholder IDs + /// provided in builder methods are used as-is. + pub fn disable_id_auto_assignment(mut self) -> Self { + self.auto_assign_ids = false; + self + } + + // --- Internal helpers --- + + fn add_field_internal(mut self, parent: Option, field: NestedFieldRef) -> Self { + self.additions.push(PendingAdd { parent, field }); + self + } +} + +// --------------------------------------------------------------------------- +// ID assignment helpers +// --------------------------------------------------------------------------- + +/// Recursively assign fresh field IDs to a `NestedField` and all its nested sub-fields. +/// +/// This follows the same recursive pattern as `ReassignFieldIds::reassign_ids_visit_type` +/// from `crate::spec::schema::id_reassigner`, but operates on new fields with placeholder +/// IDs rather than reassigning an existing schema. `ReassignFieldIds` cannot be used +/// directly here because it rejects duplicate old IDs (all new fields share placeholder +/// ID `0`). +fn assign_fresh_ids(field: &NestedField, next_id: &mut i32) -> NestedFieldRef { + *next_id += 1; + let new_id = *next_id; + let new_type = assign_fresh_ids_to_type(&field.field_type, next_id); + + Arc::new(NestedField { + id: new_id, + name: field.name.clone(), + required: field.required, + field_type: Box::new(new_type), + doc: field.doc.clone(), + initial_default: field.initial_default.clone(), + write_default: field.write_default.clone(), + }) +} + +/// Recursively assign fresh field IDs to all nested fields within a `Type`. +fn assign_fresh_ids_to_type(field_type: &Type, next_id: &mut i32) -> Type { + match field_type { + Type::Primitive(_) => field_type.clone(), + Type::Struct(struct_type) => { + let new_fields: Vec = struct_type + .fields() + .iter() + .map(|f| assign_fresh_ids(f, next_id)) + .collect(); + Type::Struct(StructType::new(new_fields)) + } + Type::List(list_type) => { + let new_element = assign_fresh_ids(&list_type.element_field, next_id); + Type::List(ListType { + element_field: new_element, + }) + } + Type::Map(map_type) => { + let new_key = assign_fresh_ids(&map_type.key_field, next_id); + let new_value = assign_fresh_ids(&map_type.value_field, next_id); + Type::Map(MapType { + key_field: new_key, + value_field: new_value, + }) + } + } +} + +// --------------------------------------------------------------------------- +// Parent path resolution +// --------------------------------------------------------------------------- + +/// Resolve a parent path to the target struct's parent field ID and a reference +/// to its `StructType`. +/// +/// If the parent is a map, navigates to the value field. If a list, navigates to +/// the element field. The final target must be a struct type. +fn resolve_parent_target<'a>( + base_schema: &'a Schema, + parent: &str, +) -> Result<(i32, &'a StructType)> { + base_schema + .field_by_name(parent) + .ok_or_else(|| { + Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot add column: parent '{parent}' not found"), + ) + }) + .and_then(|parent_field| match parent_field.field_type.as_ref() { + Type::Struct(s) => Ok((parent_field.id, s)), + Type::Map(m) => match m.value_field.field_type.as_ref() { + Type::Struct(s) => Ok((m.value_field.id, s)), + _ => Err(Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot add column: map value of '{parent}' is not a struct"), + )), + }, + Type::List(l) => match l.element_field.field_type.as_ref() { + Type::Struct(s) => Ok((l.element_field.id, s)), + _ => Err(Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot add column: list element of '{parent}' is not a struct"), + )), + }, + _ => Err(Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot add column: parent '{parent}' is not a struct, map, or list"), + )), + }) +} + +// --------------------------------------------------------------------------- +// Schema tree rebuild +// --------------------------------------------------------------------------- + +/// Rebuild a slice of fields, applying deletions and additions at every level, +/// plus any root-level additions keyed by `TABLE_ROOT_ID`. +fn rebuild_fields( + fields: &[NestedFieldRef], + adds: &HashMap>, + delete_ids: &HashSet, + root_id: i32, +) -> Vec { + fields + .iter() + .filter(|f| !delete_ids.contains(&f.id)) + .map(|f| rebuild_field(f, adds, delete_ids)) + .chain(adds.get(&root_id).into_iter().flatten().cloned()) + .collect() +} + +/// Recursively rebuild a single field. If the field (or any descendant) is a struct +/// that has pending additions, those additions are appended to the struct's fields. +/// Fields whose IDs appear in `delete_ids` are filtered out at every struct level. +fn rebuild_field( + field: &NestedFieldRef, + adds: &HashMap>, + delete_ids: &HashSet, +) -> NestedFieldRef { + match field.field_type.as_ref() { + Type::Primitive(_) => field.clone(), + Type::Struct(s) => { + let new_fields = rebuild_fields(s.fields(), adds, delete_ids, field.id); + Arc::new(NestedField { + id: field.id, + name: field.name.clone(), + required: field.required, + field_type: Box::new(Type::Struct(StructType::new(new_fields))), + doc: field.doc.clone(), + initial_default: field.initial_default.clone(), + write_default: field.write_default.clone(), + }) + } + Type::List(l) => { + let new_element = rebuild_field(&l.element_field, adds, delete_ids); + Arc::new(NestedField { + id: field.id, + name: field.name.clone(), + required: field.required, + field_type: Box::new(Type::List(ListType { + element_field: new_element, + })), + doc: field.doc.clone(), + initial_default: field.initial_default.clone(), + write_default: field.write_default.clone(), + }) + } + Type::Map(m) => { + let new_key = rebuild_field(&m.key_field, adds, delete_ids); + let new_value = rebuild_field(&m.value_field, adds, delete_ids); + Arc::new(NestedField { + id: field.id, + name: field.name.clone(), + required: field.required, + field_type: Box::new(Type::Map(MapType { + key_field: new_key, + value_field: new_value, + })), + doc: field.doc.clone(), + initial_default: field.initial_default.clone(), + write_default: field.write_default.clone(), + }) + } + } +} + +// --------------------------------------------------------------------------- +// TransactionAction implementation +// --------------------------------------------------------------------------- + +#[async_trait] +impl TransactionAction for UpdateSchemaAction { + async fn commit(self: Arc, table: &Table) -> Result { + let base_schema = table.metadata().current_schema(); + let mut last_column_id = table.metadata().last_column_id(); + + // --- 1. Validate deletes --- + let delete_ids = self + .deletes + .iter() + .map(|name: &String| { + base_schema + .field_by_name(name) + .ok_or_else(|| { + Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot delete missing column: {name}"), + ) + }) + .and_then(|field| { + match base_schema + .identifier_field_ids() + .find(|id| *id == field.id) + { + Some(_) => Err(Error::new( + ErrorKind::PreconditionFailed, + format!("Cannot delete identifier field: {name}"), + )), + None => Ok(field.id), + } + }) + }) + .collect::>>()?; + + // --- 2. Resolve parents, validate additions, assign IDs, and group by parent ID --- + // We assign IDs inline (before grouping) to preserve the caller's insertion order, + // since HashMap iteration order is non-deterministic. + let mut additions_by_parent: HashMap> = HashMap::new(); + + for pending in &self.additions { + // Check that name does not contain ".". + if pending.field.name.contains('.') { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column with ambiguous name: {}. Use the `add_column_to` method to add a column to a nested struct.", + pending.field.name + ), + )); + } + + // Required columns without an initial default need allow_incompatible_changes. + if pending.field.required && pending.field.initial_default.is_none() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Incompatible change: cannot add required column without an initial default: {}", + pending.field.name + ), + )); + } + + let parent_id = match &pending.parent { + None => { + // Root-level: check name conflict against root-level fields. + if let Some(existing) = base_schema.field_by_name(&pending.field.name) + && !delete_ids.contains(&existing.id) + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column, name already exists: {}", + pending.field.name + ), + )); + } + TABLE_ROOT_ID + } + Some(parent_path) => { + // Nested: resolve parent, check name conflict within parent struct. + let (parent_id, parent_struct) = + resolve_parent_target(base_schema, parent_path)?; + + if parent_struct + .fields() + .iter() + .any(|f| f.name == pending.field.name && !delete_ids.contains(&f.id)) + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column, name already exists in '{}': {}", + parent_path, pending.field.name + ), + )); + } + + parent_id + } + }; + + // Assign fresh IDs immediately, preserving insertion order. + let field = if self.auto_assign_ids { + assign_fresh_ids(&pending.field, &mut last_column_id) + } else { + pending.field.clone() + }; + + additions_by_parent + .entry(parent_id) + .or_default() + .push(field); + } + + // --- 4. Rebuild the schema tree with additions and deletions --- + let new_fields = rebuild_fields( + base_schema.as_struct().fields(), + &additions_by_parent, + &delete_ids, + TABLE_ROOT_ID, + ); + + // --- 5. Build the new schema --- + let schema = Schema::builder() + .with_fields(new_fields) + .with_identifier_field_ids(base_schema.identifier_field_ids()) + .build()?; + + let updates = vec![ + TableUpdate::AddSchema { schema }, + TableUpdate::SetCurrentSchema { schema_id: -1 }, + ]; + + let requirements = vec![TableRequirement::CurrentSchemaIdMatch { + current_schema_id: base_schema.schema_id(), + }]; + + Ok(ActionCommit::new(updates, requirements)) + } +} + +#[cfg(test)] +mod tests { + use std::io::BufReader; + use std::sync::Arc; + + use as_any::Downcast; + + use crate::spec::{Literal, NestedField, PrimitiveType, StructType, TableMetadata, Type}; + use crate::table::Table; + use crate::transaction::Transaction; + use crate::transaction::action::{ApplyTransactionAction, TransactionAction}; + use crate::transaction::tests::make_v2_table; + use crate::transaction::update_schema::UpdateSchemaAction; + use crate::{ErrorKind, TableIdent, TableRequirement, TableUpdate}; + + // The V2 test table has: + // last_column_id: 3 + // current schema (id=1): x(1, req, long), y(2, req, long), z(3, req, long) + // identifier_field_ids: [1, 2] + + /// Build a V2 test table that includes nested types: + /// + /// last_column_id: 14 + /// current schema (id=0): + /// x(1, req, long) -- identifier + /// y(2, req, long) -- identifier + /// z(3, req, long) + /// person(4, opt, struct) + /// name(5, opt, string) + /// age(6, req, int) + /// tags(7, opt, list) + /// element(8, req, struct) + /// key(9, opt, string) + /// value(10, opt, string) + /// props(11, opt, map) + /// key(12, req, string) + /// value(13, req, struct) + /// data(14, opt, string) + fn make_v2_table_with_nested() -> Table { + let json = r#"{ + "format-version": 2, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c2", + "location": "s3://bucket/test/location", + "last-sequence-number": 0, + "last-updated-ms": 1602638573590, + "last-column-id": 14, + "current-schema-id": 0, + "schemas": [ + { + "type": "struct", + "schema-id": 0, + "identifier-field-ids": [1, 2], + "fields": [ + {"id": 1, "name": "x", "required": true, "type": "long"}, + {"id": 2, "name": "y", "required": true, "type": "long"}, + {"id": 3, "name": "z", "required": true, "type": "long"}, + {"id": 4, "name": "person", "required": false, "type": { + "type": "struct", + "fields": [ + {"id": 5, "name": "name", "required": false, "type": "string"}, + {"id": 6, "name": "age", "required": true, "type": "int"} + ] + }}, + {"id": 7, "name": "tags", "required": false, "type": { + "type": "list", + "element-id": 8, + "element": { + "type": "struct", + "fields": [ + {"id": 9, "name": "key", "required": false, "type": "string"}, + {"id": 10, "name": "value", "required": false, "type": "string"} + ] + }, + "element-required": true + }}, + {"id": 11, "name": "props", "required": false, "type": { + "type": "map", + "key-id": 12, + "key": "string", + "value-id": 13, + "value": { + "type": "struct", + "fields": [ + {"id": 14, "name": "data", "required": false, "type": "string"} + ] + }, + "value-required": true + }} + ] + } + ], + "default-spec-id": 0, + "partition-specs": [ + {"spec-id": 0, "fields": []} + ], + "last-partition-id": 999, + "default-sort-order-id": 0, + "sort-orders": [ + {"order-id": 0, "fields": []} + ], + "properties": {}, + "current-snapshot-id": -1, + "snapshots": [] + }"#; + + let reader = BufReader::new(json.as_bytes()); + let metadata = serde_json::from_reader::<_, TableMetadata>(reader).unwrap(); + + Table::builder() + .metadata(metadata) + .metadata_location("s3://bucket/test/location/metadata/v1.json".to_string()) + .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap()) + .file_io(crate::io::FileIOBuilder::new("memory").build().unwrap()) + .build() + .unwrap() + } + + // ----------------------------------------------------------------------- + // Existing root-level tests + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn test_add_column() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let action = tx + .update_schema() + .add_column("new_col", Type::Primitive(PrimitiveType::Int)); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + let requirements = action_commit.take_requirements(); + + assert_eq!(updates.len(), 2); + + // Extract the new schema from the AddSchema update. + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + // The new field should have ID = last_column_id + 1 = 4. + let new_field = new_schema + .field_by_name("new_col") + .expect("new_col should exist"); + assert_eq!(new_field.id, 4); + assert!(!new_field.required); + assert_eq!(*new_field.field_type, Type::Primitive(PrimitiveType::Int)); + assert!(new_field.doc.is_none()); + + // Original fields should still be there. + assert!(new_schema.field_by_name("x").is_some()); + assert!(new_schema.field_by_name("y").is_some()); + assert!(new_schema.field_by_name("z").is_some()); + + assert_eq!(updates[1], TableUpdate::SetCurrentSchema { schema_id: -1 }); + + // Verify requirement. + assert_eq!(requirements.len(), 1); + assert_eq!(requirements[0], TableRequirement::CurrentSchemaIdMatch { + current_schema_id: table.metadata().current_schema().schema_id() + }); + } + + #[tokio::test] + async fn test_add_column_with_doc() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().add_column_with_doc( + "documented_col", + Type::Primitive(PrimitiveType::String), + "A documented column", + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + let field = new_schema + .field_by_name("documented_col") + .expect("documented_col should exist"); + assert_eq!(field.id, 4); + assert!(!field.required); + assert_eq!(field.doc.as_deref(), Some("A documented column")); + } + + #[tokio::test] + async fn test_add_required_column_with_initial_default() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().add_required_column( + "req_col", + Type::Primitive(PrimitiveType::Int), + Literal::int(0), + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + let field = new_schema + .field_by_name("req_col") + .expect("req_col should exist"); + assert_eq!(field.id, 4); + assert!(field.required); + assert_eq!(field.initial_default, Some(Literal::int(0))); + } + + #[tokio::test] + async fn test_add_column_name_conflict_fails() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + // "x" already exists in the V2 test schema. + let action = tx + .update_schema() + .add_column("x", Type::Primitive(PrimitiveType::Int)); + + let result = Arc::new(action).commit(&table).await; + let err = match result { + Err(e) => e, + Ok(_) => panic!("should reject adding a column with an existing name"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("already exists"), + "error should mention name conflict, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_delete_column() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + // z is not an identifier field, so we can delete it. + let action = tx.update_schema().delete_column("z"); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + assert!( + new_schema.field_by_name("z").is_none(), + "z should be deleted" + ); + assert!(new_schema.field_by_name("x").is_some()); + assert!(new_schema.field_by_name("y").is_some()); + } + + #[tokio::test] + async fn test_delete_missing_column_fails() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().delete_column("nonexistent"); + + let result = Arc::new(action).commit(&table).await; + let err = match result { + Err(e) => e, + Ok(_) => panic!("should reject deleting a non-existent column"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("nonexistent"), + "error should mention the missing column, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_add_and_delete_combined() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + // Delete z, add a new column. + let action = tx + .update_schema() + .delete_column("z") + .add_column("w", Type::Primitive(PrimitiveType::Boolean)); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + assert!( + new_schema.field_by_name("z").is_none(), + "z should be deleted" + ); + let w = new_schema.field_by_name("w").expect("w should exist"); + assert_eq!(w.id, 4); + assert!(!w.required); + } + + #[tokio::test] + async fn test_delete_and_readd_same_name() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + // Delete z, then add a new column named z -- should succeed. + let action = tx + .update_schema() + .delete_column("z") + .add_column("z", Type::Primitive(PrimitiveType::Boolean)); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + let z = new_schema + .field_by_name("z") + .expect("z should exist with new type"); + assert_eq!(z.id, 4); // new ID, not the old 3 + assert_eq!(*z.field_type, Type::Primitive(PrimitiveType::Boolean)); + } + + #[test] + fn test_apply() { + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let tx = tx + .update_schema() + .add_column("new_col", Type::Primitive(PrimitiveType::Int)) + .apply(tx) + .unwrap(); + + assert_eq!(tx.actions.len(), 1); + (*tx.actions[0]) + .downcast_ref::() + .expect("UpdateSchemaAction was not applied to Transaction!"); + } + + // ----------------------------------------------------------------------- + // Nested add tests + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn test_add_column_to_struct() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // Add "email" to the "person" struct. + let action = tx.update_schema().add_column_to( + "person", + "email", + Type::Primitive(PrimitiveType::String), + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + // "email" should be nested under "person" with ID = last_column_id + 1 = 15. + let email = new_schema + .field_by_name("person.email") + .expect("person.email should exist"); + assert_eq!(email.id, 15); + assert!(!email.required); + assert_eq!(*email.field_type, Type::Primitive(PrimitiveType::String)); + + // Original nested fields should still be there. + assert!(new_schema.field_by_name("person.name").is_some()); + assert!(new_schema.field_by_name("person.age").is_some()); + } + + #[tokio::test] + async fn test_add_column_to_struct_with_doc() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().add_column_to_with_doc( + "person", + "phone", + Type::Primitive(PrimitiveType::String), + "Phone number", + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + let phone = new_schema + .field_by_name("person.phone") + .expect("person.phone should exist"); + assert_eq!(phone.id, 15); + assert_eq!(phone.doc.as_deref(), Some("Phone number")); + } + + #[tokio::test] + async fn test_add_column_to_list_element_struct() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // "tags" is a list. Adding to the list navigates to its + // element struct automatically. + let action = tx.update_schema().add_column_to( + "tags", + "score", + Type::Primitive(PrimitiveType::Double), + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + // The list element struct should now contain "score". + let score = new_schema + .field_by_name("tags.element.score") + .expect("tags.element.score should exist"); + assert_eq!(score.id, 15); + assert!(!score.required); + + // Existing fields preserved. + assert!(new_schema.field_by_name("tags.element.key").is_some()); + assert!(new_schema.field_by_name("tags.element.value").is_some()); + } + + #[tokio::test] + async fn test_add_column_to_map_value_struct() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // "props" is a map. Adding to the map navigates to its + // value struct automatically. + let action = tx.update_schema().add_column_to( + "props", + "version", + Type::Primitive(PrimitiveType::Int), + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + let version = new_schema + .field_by_name("props.value.version") + .expect("props.value.version should exist"); + assert_eq!(version.id, 15); + + // Existing map value fields preserved. + assert!(new_schema.field_by_name("props.value.data").is_some()); + } + + #[tokio::test] + async fn test_add_column_to_nonexistent_parent_fails() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().add_column_to( + "nonexistent", + "col", + Type::Primitive(PrimitiveType::Int), + ); + + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("should reject adding to a nonexistent parent"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("nonexistent"), + "error should mention the missing parent, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_add_column_to_primitive_parent_fails() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // "x" is a primitive (long), not a struct. + let action = + tx.update_schema() + .add_column_to("x", "col", Type::Primitive(PrimitiveType::Int)); + + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("should reject adding to a primitive parent"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("not a struct"), + "error should mention type mismatch, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_add_column_to_nested_name_conflict_fails() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // "name" already exists in the "person" struct. + let action = tx.update_schema().add_column_to( + "person", + "name", + Type::Primitive(PrimitiveType::String), + ); + + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("should reject adding a column with conflicting name"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("already exists"), + "error should mention name conflict, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_root_and_nested_add_combined() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // Add a root column and a nested column in the same action. + let action = tx + .update_schema() + .add_column("root_col", Type::Primitive(PrimitiveType::Boolean)) + .add_column_to("person", "email", Type::Primitive(PrimitiveType::String)); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + // Root column gets the first fresh ID. + let root_col = new_schema + .field_by_name("root_col") + .expect("root_col should exist"); + assert_eq!(root_col.id, 15); + + // Nested column gets the next ID. + let email = new_schema + .field_by_name("person.email") + .expect("person.email should exist"); + assert_eq!(email.id, 16); + } + + #[tokio::test] + async fn test_add_nested_struct_type_with_fresh_ids() { + // Exercises the assign_fresh_ids bug fix: adding a new column whose TYPE + // contains nested fields (e.g. a struct column). All sub-fields must receive + // fresh IDs, not placeholder 0. + let table = make_v2_table(); + let tx = Transaction::new(&table); + + let action = tx.update_schema().add_column( + "address", + Type::Struct(StructType::new(vec![ + NestedField::optional(0, "street", Type::Primitive(PrimitiveType::String)).into(), + NestedField::optional(0, "city", Type::Primitive(PrimitiveType::String)).into(), + ])), + ); + + let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); + let updates = action_commit.take_updates(); + + let new_schema = match &updates[0] { + TableUpdate::AddSchema { schema } => schema, + other => panic!("expected AddSchema, got {:?}", other), + }; + + // "address" gets ID 4 (last_column_id=3, +1). + let address = new_schema + .field_by_name("address") + .expect("address should exist"); + assert_eq!(address.id, 4); + + // Sub-fields get IDs 5 and 6. + let street = new_schema + .field_by_name("address.street") + .expect("address.street should exist"); + assert_eq!(street.id, 5); + + let city = new_schema + .field_by_name("address.city") + .expect("address.city should exist"); + assert_eq!(city.id, 6); + } +} diff --git a/crates/integration_tests/tests/update_schema.rs b/crates/integration_tests/tests/update_schema.rs new file mode 100644 index 0000000000..87a1d63f16 --- /dev/null +++ b/crates/integration_tests/tests/update_schema.rs @@ -0,0 +1,409 @@ +// 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. + +//! Integration tests for the `UpdateSchemaAction`. + +mod common; + +use std::sync::Arc; + +use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray, StructArray}; +use common::{random_ns, test_schema}; +use futures::TryStreamExt; +use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; +use iceberg::writer::file_writer::ParquetWriterBuilder; +use iceberg::writer::file_writer::location_generator::{ + DefaultFileNameGenerator, DefaultLocationGenerator, +}; +use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; +use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; +use iceberg::{Catalog, CatalogBuilder, TableCreation}; +use iceberg_catalog_rest::RestCatalogBuilder; +use iceberg_integration_tests::get_test_fixture; +use parquet::arrow::arrow_reader::ArrowReaderOptions; +use parquet::file::properties::WriterProperties; + +/// Creates a table, appends data, adds a new field to the schema, +/// verifies existing data is still readable, then appends data with the new schema. +#[tokio::test] +async fn test_add_field() { + let fixture = get_test_fixture(); + let rest_catalog = RestCatalogBuilder::default() + .load("rest", fixture.catalog_config.clone()) + .await + .unwrap(); + let ns = random_ns().await; + let schema = test_schema(); + + let table_creation = TableCreation::builder() + .name("t1".to_string()) + .schema(schema.clone()) + .build(); + + let table = rest_catalog + .create_table(ns.name(), table_creation) + .await + .unwrap(); + + // Create the writer and write initial data + let arrow_schema: Arc = Arc::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + let location_generator = DefaultLocationGenerator::new(table.metadata().clone()).unwrap(); + let file_name_generator = DefaultFileNameGenerator::new( + "test".to_string(), + None, + iceberg::spec::DataFileFormat::Parquet, + ); + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); + let col1 = StringArray::from(vec![Some("foo"), Some("bar"), None, Some("baz")]); + let col2 = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(4)]); + let col3 = BooleanArray::from(vec![Some(true), Some(false), None, Some(false)]); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![ + Arc::new(col1) as ArrayRef, + Arc::new(col2) as ArrayRef, + Arc::new(col3) as ArrayRef, + ]) + .unwrap(); + data_file_writer.write(batch.clone()).await.unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + // Check parquet file schema has the expected field IDs + let content = table + .file_io() + .new_input(data_file[0].file_path()) + .unwrap() + .read() + .await + .unwrap(); + let parquet_reader = parquet::arrow::arrow_reader::ArrowReaderMetadata::load( + &content, + ArrowReaderOptions::default(), + ) + .unwrap(); + let field_ids: Vec = parquet_reader + .parquet_schema() + .columns() + .iter() + .map(|col| col.self_type().get_basic_info().id()) + .collect(); + assert_eq!(field_ids, vec![1, 2, 3]); + + // Commit the initial data + let tx = Transaction::new(&table); + let append_action = tx.fast_append().add_data_files(data_file.clone()); + let tx = append_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Verify the initial data is readable + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0], batch); + + // Add a new optional primitive field to the table + let tx = Transaction::new(&table); + let add_action = tx.update_schema().add_column( + "a", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::Int), + ); + let tx = add_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Verify existing data is still readable after schema evolution + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + assert_eq!(batches[0], batch); + + // Add a struct column, then add a nested column inside it + let tx = Transaction::new(&table); + let add_action = tx.update_schema().add_column( + "info", + iceberg::spec::Type::Struct(iceberg::spec::StructType::new(vec![Arc::new( + iceberg::spec::NestedField::optional( + 0, + "city", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), + ), + )])), + ); + let tx = add_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Verify the struct column was added + let schema = table.metadata().current_schema(); + let info_field = schema.field_by_name("info").expect("info field should exist"); + assert!(matches!(info_field.field_type.as_ref(), iceberg::spec::Type::Struct(_))); + let city_field = schema + .field_by_name("info.city") + .expect("info.city field should exist"); + assert!(matches!( + city_field.field_type.as_ref(), + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String) + )); + + // Add a nested column to the struct + let tx = Transaction::new(&table); + let add_action = tx.update_schema().add_column_to( + "info", + "zip", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), + ); + let tx = add_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Verify the nested column was added + let schema = table.metadata().current_schema(); + let zip_field = schema + .field_by_name("info.zip") + .expect("info.zip field should exist"); + assert!(matches!( + zip_field.field_type.as_ref(), + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String) + )); + + // Verify existing data is still readable + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + + // Create a new writer with the evolved schema and write data including the new field + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); + let col1 = StringArray::from(vec![Some("foo"), Some("bar"), None, Some("baz")]); + let col2 = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(4)]); + let col3 = BooleanArray::from(vec![Some(true), Some(false), None, Some(false)]); + let col4 = Int32Array::from(vec![Some(1), Some(2), Some(3), Some(4)]); + let evolved_arrow_schema: Arc = Arc::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + // Build a struct array for the "info" column: {city, zip} + let city_array = StringArray::from(vec![Some("NYC"), Some("LA"), None, Some("SF")]); + let zip_array = StringArray::from(vec![Some("10001"), None, Some("90001"), Some("94101")]); + let info_fields = evolved_arrow_schema + .field_with_name("info") + .unwrap() + .data_type() + .clone(); + let struct_fields = match &info_fields { + arrow_schema::DataType::Struct(fields) => fields.clone(), + _ => panic!("expected struct type for info"), + }; + let info_array = StructArray::try_new( + struct_fields, + vec![ + Arc::new(city_array) as ArrayRef, + Arc::new(zip_array) as ArrayRef, + ], + None, + ) + .unwrap(); + let batch_with_new_field = RecordBatch::try_new(evolved_arrow_schema.clone(), vec![ + Arc::new(col1) as ArrayRef, + Arc::new(col2) as ArrayRef, + Arc::new(col3) as ArrayRef, + Arc::new(col4) as ArrayRef, + Arc::new(info_array) as ArrayRef, + ]) + .unwrap(); + data_file_writer + .write(batch_with_new_field.clone()) + .await + .unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + // Commit the new data with evolved schema + let tx = Transaction::new(&table); + let append_action = tx.fast_append().add_data_files(data_file.clone()); + let tx = append_action.apply(tx).unwrap(); + let _table = tx.commit(&rest_catalog).await.unwrap(); +} + +/// Creates a table, adds data, deletes a non-identifier column, +/// and verifies the schema was updated and existing data is still readable. +#[tokio::test] +async fn test_delete_field() { + let fixture = get_test_fixture(); + let rest_catalog = RestCatalogBuilder::default() + .load("rest", fixture.catalog_config.clone()) + .await + .unwrap(); + let ns = random_ns().await; + let schema = test_schema(); + + let table_creation = TableCreation::builder() + .name("t_delete".to_string()) + .schema(schema.clone()) + .build(); + + let table = rest_catalog + .create_table(ns.name(), table_creation) + .await + .unwrap(); + + // Write initial data with all three columns + let arrow_schema: Arc = Arc::new( + table + .metadata() + .current_schema() + .as_ref() + .try_into() + .unwrap(), + ); + let location_generator = DefaultLocationGenerator::new(table.metadata().clone()).unwrap(); + let file_name_generator = DefaultFileNameGenerator::new( + "test".to_string(), + None, + iceberg::spec::DataFileFormat::Parquet, + ); + let parquet_writer_builder = ParquetWriterBuilder::new( + WriterProperties::default(), + table.metadata().current_schema().clone(), + ); + let rolling_file_writer_builder = RollingFileWriterBuilder::new_with_default_file_size( + parquet_writer_builder, + table.file_io().clone(), + location_generator.clone(), + file_name_generator.clone(), + ); + let data_file_writer_builder = DataFileWriterBuilder::new(rolling_file_writer_builder); + let mut data_file_writer = data_file_writer_builder.build(None).await.unwrap(); + let col1 = StringArray::from(vec![Some("foo"), Some("bar")]); + let col2 = Int32Array::from(vec![Some(1), Some(2)]); + let col3 = BooleanArray::from(vec![Some(true), Some(false)]); + let batch = RecordBatch::try_new(arrow_schema.clone(), vec![ + Arc::new(col1) as ArrayRef, + Arc::new(col2) as ArrayRef, + Arc::new(col3) as ArrayRef, + ]) + .unwrap(); + data_file_writer.write(batch.clone()).await.unwrap(); + let data_file = data_file_writer.close().await.unwrap(); + + // Commit the initial data + let tx = Transaction::new(&table); + let append_action = tx.fast_append().add_data_files(data_file.clone()); + let tx = append_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Delete the optional "baz" column (field 3, not an identifier) + let tx = Transaction::new(&table); + let delete_action = tx.update_schema().delete_column("baz"); + let tx = delete_action.apply(tx).unwrap(); + let table = tx.commit(&rest_catalog).await.unwrap(); + + // Verify the schema no longer contains "baz" + let schema = table.metadata().current_schema(); + assert!( + schema.field_by_name("baz").is_none(), + "baz should have been deleted" + ); + assert!( + schema.field_by_name("foo").is_some(), + "foo should still exist" + ); + assert!( + schema.field_by_name("bar").is_some(), + "bar should still exist" + ); + + // Verify existing data is still readable after the column deletion + let batch_stream = table + .scan() + .select_all() + .build() + .unwrap() + .to_arrow() + .await + .unwrap(); + let batches: Vec<_> = batch_stream.try_collect().await.unwrap(); + assert_eq!(batches.len(), 1); + + // Deleting an identifier field should fail + let tx = Transaction::new(&table); + let delete_action = tx.update_schema().delete_column("bar"); + let tx = delete_action.apply(tx).unwrap(); + let result = tx.commit(&rest_catalog).await; + assert!( + result.is_err(), + "deleting an identifier field should fail" + ); + + // Deleting a non-existent column should fail + let tx = Transaction::new(&table); + let delete_action = tx.update_schema().delete_column("nonexistent"); + let tx = delete_action.apply(tx).unwrap(); + let result = tx.commit(&rest_catalog).await; + assert!( + result.is_err(), + "deleting a non-existent column should fail" + ); +} From 6afe97e6408c2c7239ad89bd45ebf40b8ac11bdd Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Tue, 10 Feb 2026 14:41:36 +0200 Subject: [PATCH 02/10] Cargo format update --- crates/integration_tests/tests/update_schema.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/crates/integration_tests/tests/update_schema.rs b/crates/integration_tests/tests/update_schema.rs index 87a1d63f16..789a55ad69 100644 --- a/crates/integration_tests/tests/update_schema.rs +++ b/crates/integration_tests/tests/update_schema.rs @@ -178,8 +178,13 @@ async fn test_add_field() { // Verify the struct column was added let schema = table.metadata().current_schema(); - let info_field = schema.field_by_name("info").expect("info field should exist"); - assert!(matches!(info_field.field_type.as_ref(), iceberg::spec::Type::Struct(_))); + let info_field = schema + .field_by_name("info") + .expect("info field should exist"); + assert!(matches!( + info_field.field_type.as_ref(), + iceberg::spec::Type::Struct(_) + )); let city_field = schema .field_by_name("info.city") .expect("info.city field should exist"); @@ -392,10 +397,7 @@ async fn test_delete_field() { let delete_action = tx.update_schema().delete_column("bar"); let tx = delete_action.apply(tx).unwrap(); let result = tx.commit(&rest_catalog).await; - assert!( - result.is_err(), - "deleting an identifier field should fail" - ); + assert!(result.is_err(), "deleting an identifier field should fail"); // Deleting a non-existent column should fail let tx = Transaction::new(&table); From 9a537eb62557d82b84afdd91bb82e2c1e4ba9ad1 Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Tue, 10 Feb 2026 14:52:06 +0200 Subject: [PATCH 03/10] Fix clippy! --- .../iceberg/src/transaction/update_schema.rs | 24 +++++++++---------- 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index 3caa275be1..c828d4e438 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -703,7 +703,7 @@ mod tests { // Extract the new schema from the AddSchema update. let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; // The new field should have ID = last_column_id + 1 = 4. @@ -745,7 +745,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; let field = new_schema @@ -772,7 +772,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; let field = new_schema @@ -819,7 +819,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; assert!( @@ -866,7 +866,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; assert!( @@ -894,7 +894,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; let z = new_schema @@ -942,7 +942,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; // "email" should be nested under "person" with ID = last_column_id + 1 = 15. @@ -975,7 +975,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; let phone = new_schema @@ -1003,7 +1003,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; // The list element struct should now contain "score". @@ -1036,7 +1036,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; let version = new_schema @@ -1133,7 +1133,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; // Root column gets the first fresh ID. @@ -1170,7 +1170,7 @@ mod tests { let new_schema = match &updates[0] { TableUpdate::AddSchema { schema } => schema, - other => panic!("expected AddSchema, got {:?}", other), + other => panic!("expected AddSchema, got {other:?}"), }; // "address" gets ID 4 (last_column_id=3, +1). From 2b12d5db2583613e43c746a3c035622aacf5f464 Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Mon, 9 Mar 2026 14:32:53 +0200 Subject: [PATCH 04/10] Add write_default to new fields --- crates/iceberg/src/transaction/update_schema.rs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index c828d4e438..33a4fc2dc5 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -114,7 +114,9 @@ impl UpdateSchemaAction { initial_default: Literal, ) -> Self { self.add_field(Arc::new( - NestedField::required(0, name, field_type).with_initial_default(initial_default), + NestedField::required(0, name, field_type) + .with_initial_default(initial_default.clone()) + .with_write_default(initial_default), )) } @@ -132,7 +134,8 @@ impl UpdateSchemaAction { ) -> Self { self.add_field(Arc::new( NestedField::required(0, name, field_type) - .with_initial_default(initial_default) + .with_initial_default(initial_default.clone()) + .with_write_default(initial_default) .with_doc(doc), )) } @@ -192,7 +195,9 @@ impl UpdateSchemaAction { self.add_field_to( parent, Arc::new( - NestedField::required(0, name, field_type).with_initial_default(initial_default), + NestedField::required(0, name, field_type) + .with_initial_default(initial_default.clone()) + .with_write_default(initial_default), ), ) } @@ -213,7 +218,8 @@ impl UpdateSchemaAction { parent, Arc::new( NestedField::required(0, name, field_type) - .with_initial_default(initial_default) + .with_initial_default(initial_default.clone()) + .with_write_default(initial_default) .with_doc(doc), ), ) @@ -781,6 +787,7 @@ mod tests { assert_eq!(field.id, 4); assert!(field.required); assert_eq!(field.initial_default, Some(Literal::int(0))); + assert_eq!(field.write_default, Some(Literal::int(0))); } #[tokio::test] From c221e878a4b66427538ef0e25c1da4ffbde09c67 Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Mon, 9 Mar 2026 14:55:56 +0200 Subject: [PATCH 05/10] Fix new FileIO api --- crates/iceberg/src/transaction/update_schema.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index 33a4fc2dc5..b1ce9f3496 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -682,7 +682,7 @@ mod tests { .metadata(metadata) .metadata_location("s3://bucket/test/location/metadata/v1.json".to_string()) .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap()) - .file_io(crate::io::FileIOBuilder::new("memory").build().unwrap()) + .file_io(crate::io::FileIO::new_with_memory()) .build() .unwrap() } From c0f8342f23aae1039ba205cebdd2c4d07fcaa52d Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Mon, 9 Mar 2026 15:13:16 +0200 Subject: [PATCH 06/10] Add integration test storage factory --- crates/integration_tests/tests/update_schema.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/integration_tests/tests/update_schema.rs b/crates/integration_tests/tests/update_schema.rs index 789a55ad69..17b106f669 100644 --- a/crates/integration_tests/tests/update_schema.rs +++ b/crates/integration_tests/tests/update_schema.rs @@ -35,6 +35,7 @@ use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; use iceberg::{Catalog, CatalogBuilder, TableCreation}; use iceberg_catalog_rest::RestCatalogBuilder; use iceberg_integration_tests::get_test_fixture; +use iceberg_storage_opendal::OpenDalStorageFactory; use parquet::arrow::arrow_reader::ArrowReaderOptions; use parquet::file::properties::WriterProperties; @@ -44,6 +45,10 @@ use parquet::file::properties::WriterProperties; async fn test_add_field() { let fixture = get_test_fixture(); let rest_catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { + configured_scheme: "s3".to_string(), + customized_credential_load: None, + })) .load("rest", fixture.catalog_config.clone()) .await .unwrap(); @@ -298,6 +303,10 @@ async fn test_add_field() { async fn test_delete_field() { let fixture = get_test_fixture(); let rest_catalog = RestCatalogBuilder::default() + .with_storage_factory(Arc::new(OpenDalStorageFactory::S3 { + configured_scheme: "s3".to_string(), + customized_credential_load: None, + })) .load("rest", fixture.catalog_config.clone()) .await .unwrap(); From 85a56d93bd03ff538eea6335cfa86e39ea4691ff Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Mon, 16 Mar 2026 14:09:28 +0200 Subject: [PATCH 07/10] add extra delete check for parent --- crates/iceberg/src/transaction/update_schema.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index b1ce9f3496..b8c9845e59 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -512,11 +512,11 @@ impl TransactionAction for UpdateSchemaAction { let (parent_id, parent_struct) = resolve_parent_target(base_schema, parent_path)?; - if parent_struct - .fields() - .iter() - .any(|f| f.name == pending.field.name && !delete_ids.contains(&f.id)) - { + if parent_struct.fields().iter().any(|f| { + f.name == pending.field.name + && !delete_ids.contains(&f.id) + && !delete_ids.contains(&parent_id) + }) { return Err(Error::new( ErrorKind::PreconditionFailed, format!( From 17e900c8df3e4c543ace25e3131f07c365d81142 Mon Sep 17 00:00:00 2001 From: Victor Ghita Date: Mon, 16 Mar 2026 15:59:20 +0200 Subject: [PATCH 08/10] Refactor AddColumn API --- crates/iceberg/src/transaction/mod.rs | 1 + .../iceberg/src/transaction/update_schema.rs | 391 ++++++++---------- .../integration_tests/tests/update_schema.rs | 20 +- 3 files changed, 176 insertions(+), 236 deletions(-) diff --git a/crates/iceberg/src/transaction/mod.rs b/crates/iceberg/src/transaction/mod.rs index 9bf569ac95..630689e5d8 100644 --- a/crates/iceberg/src/transaction/mod.rs +++ b/crates/iceberg/src/transaction/mod.rs @@ -67,6 +67,7 @@ use std::sync::Arc; use std::time::Duration; use backon::{BackoffBuilder, ExponentialBackoff, ExponentialBuilder, RetryableWithContext}; +pub use update_schema::AddColumn; use crate::error::Result; use crate::spec::TableProperties; diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index b8c9845e59..5d0b435d8d 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -19,6 +19,7 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; use async_trait::async_trait; +use typed_builder::TypedBuilder; use crate::spec::{ ListType, Literal, MapType, NestedField, NestedFieldRef, Schema, StructType, Type, @@ -29,14 +30,78 @@ use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate}; /// Sentinel parent ID representing the table root (top-level columns). const TABLE_ROOT_ID: i32 = -1; +// Default ID for a new column. This will be re-assigned to a fresh ID at commit time. +const DEFAULT_ID: i32 = 0; -/// A pending column addition, recording the parent path and the field to add. -struct PendingAdd { - /// `None` means a root-level addition; `Some("person")` or `Some("person.address")` - /// identifies the nested struct to add the column to. +#[derive(TypedBuilder)] +/// Declarative specification for adding a column in [`UpdateSchemaAction`]. +/// +/// Use helper constructors such as [`AddColumn::optional`] and [`AddColumn::required`], +/// optionally combined with [`AddColumn::with_parent`] and [`AddColumn::with_doc`], then pass +/// the value to +/// [`UpdateSchemaAction::add_column`]. +pub struct AddColumn { + #[builder(default = None, setter(strip_option, into))] parent: Option, - /// The field to add. Uses placeholder ID `0` which is auto-assigned at commit time. - field: NestedFieldRef, + #[builder(setter(into))] + name: String, + #[builder(default = false)] + required: bool, + field_type: Type, + #[builder(default = None, setter(strip_option, into))] + doc: Option, + #[builder(default = None, setter(strip_option))] + initial_default: Option, + #[builder(default = None, setter(strip_option))] + write_default: Option, +} + +impl AddColumn { + /// Create a root-level optional column specification. + pub fn optional(name: impl ToString, field_type: Type) -> Self { + Self::builder() + .name(name.to_string()) + .field_type(field_type) + .required(false) + .build() + } + + /// Create a root-level required column specification. + pub fn required(name: impl ToString, field_type: Type, initial_default: Literal) -> Self { + Self::builder() + .name(name.to_string()) + .field_type(field_type) + .required(true) + .initial_default(initial_default.clone()) + .write_default(initial_default) + .build() + } + + /// Return a copy with an updated parent path. + pub fn with_parent(mut self, parent: impl ToString) -> Self { + self.parent = Some(parent.to_string()); + self + } + + /// Return a copy with an updated doc string. + pub fn with_doc(mut self, doc: impl ToString) -> Self { + self.doc = Some(doc.to_string()); + self + } + + fn to_nested_field(&self) -> NestedFieldRef { + let mut field = NestedField::new( + DEFAULT_ID, + self.name.clone(), + self.field_type.clone(), + self.required, + ); + + field.doc = self.doc.clone(); + field.initial_default = self.initial_default.clone(); + field.write_default = self.write_default.clone(); + Arc::new(field) + } } /// Schema evolution API modeled after the Java `SchemaUpdate` implementation. @@ -52,14 +117,17 @@ struct PendingAdd { /// ```ignore /// let tx = Transaction::new(&table); /// let action = tx.update_schema() -/// .add_column("new_col", Type::Primitive(PrimitiveType::Int)) -/// .add_column_to("person", "email", Type::Primitive(PrimitiveType::String)) +/// .add_column(AddColumn::optional("new_col", Type::Primitive(PrimitiveType::Int))) +/// .add_column( +/// AddColumn::optional("email", Type::Primitive(PrimitiveType::String)) +/// .with_parent("person") +/// ) /// .delete_column("old_col"); /// let tx = action.apply(tx).unwrap(); /// let table = tx.commit(&catalog).await.unwrap(); /// ``` pub struct UpdateSchemaAction { - additions: Vec, + additions: Vec, deletes: Vec, auto_assign_ids: bool, } @@ -76,153 +144,14 @@ impl UpdateSchemaAction { // --- Root-level additions --- - /// Add a `NestedFieldRef` column to the table root. - pub fn add_field(self, field: NestedFieldRef) -> Self { - self.add_field_internal(None, field) - } - - /// Add an optional column to the table root. - /// - /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. - pub fn add_column(self, name: impl ToString, field_type: Type) -> Self { - self.add_field(Arc::new(NestedField::optional(0, name, field_type))) - } - - /// Add an optional column with a doc string to the table root. - /// - /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. - pub fn add_column_with_doc( - self, - name: impl ToString, - field_type: Type, - doc: impl ToString, - ) -> Self { - self.add_field(Arc::new( - NestedField::optional(0, name, field_type).with_doc(doc), - )) - } - - /// Add a required column to the table root. - /// - /// An `initial_default` value is required per the Iceberg spec: it is used to populate - /// this field for all records that were written before the field was added. - /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. - pub fn add_required_column( - self, - name: impl ToString, - field_type: Type, - initial_default: Literal, - ) -> Self { - self.add_field(Arc::new( - NestedField::required(0, name, field_type) - .with_initial_default(initial_default.clone()) - .with_write_default(initial_default), - )) - } - - /// Add a required column with a doc string to the table root. - /// - /// An `initial_default` value is required per the Iceberg spec: it is used to populate - /// this field for all records that were written before the field was added. - /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. - pub fn add_required_column_with_doc( - self, - name: impl ToString, - field_type: Type, - initial_default: Literal, - doc: impl ToString, - ) -> Self { - self.add_field(Arc::new( - NestedField::required(0, name, field_type) - .with_initial_default(initial_default.clone()) - .with_write_default(initial_default) - .with_doc(doc), - )) - } - - // --- Nested additions --- - - /// Add a `NestedFieldRef` column under a parent struct identified by name. - /// - /// If the parent is a map, the column is added to the map value's struct. - /// If the parent is a list, the column is added to the list element's struct. - pub fn add_field_to(self, parent: impl ToString, field: NestedFieldRef) -> Self { - self.add_field_internal(Some(parent.to_string()), field) - } - - /// Add an optional column under a parent struct identified by name. - /// - /// The `parent` can be a dotted path (e.g. `"person"` or `"person.address"`). - /// If the parent is a map, the column is added to the map value's struct. - /// If the parent is a list, the column is added to the list element's struct. - /// The field ID is a placeholder (`0`) and will be auto-assigned at commit time. - pub fn add_column_to( - self, - parent: impl ToString, - name: impl ToString, - field_type: Type, - ) -> Self { - self.add_field_to(parent, Arc::new(NestedField::optional(0, name, field_type))) - } - - /// Add an optional column with a doc string under a parent struct. - /// - /// See [`add_column_to`](Self::add_column_to) for parent path details. - pub fn add_column_to_with_doc( - self, - parent: impl ToString, - name: impl ToString, - field_type: Type, - doc: impl ToString, - ) -> Self { - self.add_field_to( - parent, - Arc::new(NestedField::optional(0, name, field_type).with_doc(doc)), - ) - } - - /// Add a required column under a parent struct. - /// - /// See [`add_column_to`](Self::add_column_to) for parent path details. - /// An `initial_default` value is required per the Iceberg spec. - pub fn add_required_column_to( - self, - parent: impl ToString, - name: impl ToString, - field_type: Type, - initial_default: Literal, - ) -> Self { - self.add_field_to( - parent, - Arc::new( - NestedField::required(0, name, field_type) - .with_initial_default(initial_default.clone()) - .with_write_default(initial_default), - ), - ) - } - - /// Add a required column with a doc string under a parent struct. + /// Add a column to the table schema. /// - /// See [`add_column_to`](Self::add_column_to) for parent path details. - /// An `initial_default` value is required per the Iceberg spec. - pub fn add_required_column_to_with_doc( - self, - parent: impl ToString, - name: impl ToString, - field_type: Type, - initial_default: Literal, - doc: impl ToString, - ) -> Self { - self.add_field_to( - parent, - Arc::new( - NestedField::required(0, name, field_type) - .with_initial_default(initial_default.clone()) - .with_write_default(initial_default) - .with_doc(doc), - ), - ) + /// To add a root-level column, leave `AddColumn::parent` as `None`. + /// For nested additions, set a parent path (for example via [`AddColumn::with_parent`]). + /// If the parent resolves to a map/list, the column is added to map value/list element. + pub fn add_column(mut self, add_column: AddColumn) -> Self { + self.additions.push(add_column); + self } // --- Other builder methods --- @@ -243,11 +172,6 @@ impl UpdateSchemaAction { } // --- Internal helpers --- - - fn add_field_internal(mut self, parent: Option, field: NestedFieldRef) -> Self { - self.additions.push(PendingAdd { parent, field }); - self - } } // --------------------------------------------------------------------------- @@ -260,7 +184,7 @@ impl UpdateSchemaAction { /// from `crate::spec::schema::id_reassigner`, but operates on new fields with placeholder /// IDs rather than reassigning an existing schema. `ReassignFieldIds` cannot be used /// directly here because it rejects duplicate old IDs (all new fields share placeholder -/// ID `0`). +/// ID `DEFAULT_ID`). fn assign_fresh_ids(field: &NestedField, next_id: &mut i32) -> NestedFieldRef { *next_id += 1; let new_id = *next_id; @@ -468,40 +392,42 @@ impl TransactionAction for UpdateSchemaAction { // since HashMap iteration order is non-deterministic. let mut additions_by_parent: HashMap> = HashMap::new(); - for pending in &self.additions { + for add in &self.additions { + let pending_field = add.to_nested_field(); + // Check that name does not contain ".". - if pending.field.name.contains('.') { + if pending_field.name.contains('.') { return Err(Error::new( ErrorKind::PreconditionFailed, format!( - "Cannot add column with ambiguous name: {}. Use the `add_column_to` method to add a column to a nested struct.", - pending.field.name + "Cannot add column with ambiguous name: {}. Use `AddColumn::with_parent` to add a column to a nested struct.", + pending_field.name ), )); } // Required columns without an initial default need allow_incompatible_changes. - if pending.field.required && pending.field.initial_default.is_none() { + if pending_field.required && pending_field.initial_default.is_none() { return Err(Error::new( ErrorKind::PreconditionFailed, format!( "Incompatible change: cannot add required column without an initial default: {}", - pending.field.name + pending_field.name ), )); } - let parent_id = match &pending.parent { + let parent_id = match &add.parent { None => { // Root-level: check name conflict against root-level fields. - if let Some(existing) = base_schema.field_by_name(&pending.field.name) + if let Some(existing) = base_schema.field_by_name(&pending_field.name) && !delete_ids.contains(&existing.id) { return Err(Error::new( ErrorKind::PreconditionFailed, format!( "Cannot add column, name already exists: {}", - pending.field.name + pending_field.name ), )); } @@ -513,7 +439,7 @@ impl TransactionAction for UpdateSchemaAction { resolve_parent_target(base_schema, parent_path)?; if parent_struct.fields().iter().any(|f| { - f.name == pending.field.name + f.name == pending_field.name && !delete_ids.contains(&f.id) && !delete_ids.contains(&parent_id) }) { @@ -521,7 +447,7 @@ impl TransactionAction for UpdateSchemaAction { ErrorKind::PreconditionFailed, format!( "Cannot add column, name already exists in '{}': {}", - parent_path, pending.field.name + parent_path, pending_field.name ), )); } @@ -532,9 +458,9 @@ impl TransactionAction for UpdateSchemaAction { // Assign fresh IDs immediately, preserving insertion order. let field = if self.auto_assign_ids { - assign_fresh_ids(&pending.field, &mut last_column_id) + assign_fresh_ids(&pending_field, &mut last_column_id) } else { - pending.field.clone() + pending_field }; additions_by_parent @@ -582,7 +508,7 @@ mod tests { use crate::transaction::Transaction; use crate::transaction::action::{ApplyTransactionAction, TransactionAction}; use crate::transaction::tests::make_v2_table; - use crate::transaction::update_schema::UpdateSchemaAction; + use crate::transaction::update_schema::{AddColumn, DEFAULT_ID, UpdateSchemaAction}; use crate::{ErrorKind, TableIdent, TableRequirement, TableUpdate}; // The V2 test table has: @@ -696,9 +622,10 @@ mod tests { let table = make_v2_table(); let tx = Transaction::new(&table); - let action = tx - .update_schema() - .add_column("new_col", Type::Primitive(PrimitiveType::Int)); + let action = tx.update_schema().add_column(AddColumn::optional( + "new_col", + Type::Primitive(PrimitiveType::Int), + )); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); @@ -740,10 +667,9 @@ mod tests { let table = make_v2_table(); let tx = Transaction::new(&table); - let action = tx.update_schema().add_column_with_doc( - "documented_col", - Type::Primitive(PrimitiveType::String), - "A documented column", + let action = tx.update_schema().add_column( + AddColumn::optional("documented_col", Type::Primitive(PrimitiveType::String)) + .with_doc("A documented column"), ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); @@ -767,11 +693,11 @@ mod tests { let table = make_v2_table(); let tx = Transaction::new(&table); - let action = tx.update_schema().add_required_column( + let action = tx.update_schema().add_column(AddColumn::required( "req_col", Type::Primitive(PrimitiveType::Int), Literal::int(0), - ); + )); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); @@ -796,9 +722,10 @@ mod tests { let tx = Transaction::new(&table); // "x" already exists in the V2 test schema. - let action = tx - .update_schema() - .add_column("x", Type::Primitive(PrimitiveType::Int)); + let action = tx.update_schema().add_column(AddColumn::optional( + "x", + Type::Primitive(PrimitiveType::Int), + )); let result = Arc::new(action).commit(&table).await; let err = match result { @@ -866,7 +793,10 @@ mod tests { let action = tx .update_schema() .delete_column("z") - .add_column("w", Type::Primitive(PrimitiveType::Boolean)); + .add_column(AddColumn::optional( + "w", + Type::Primitive(PrimitiveType::Boolean), + )); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); @@ -894,7 +824,10 @@ mod tests { let action = tx .update_schema() .delete_column("z") - .add_column("z", Type::Primitive(PrimitiveType::Boolean)); + .add_column(AddColumn::optional( + "z", + Type::Primitive(PrimitiveType::Boolean), + )); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); @@ -918,7 +851,10 @@ mod tests { let tx = tx .update_schema() - .add_column("new_col", Type::Primitive(PrimitiveType::Int)) + .add_column(AddColumn::optional( + "new_col", + Type::Primitive(PrimitiveType::Int), + )) .apply(tx) .unwrap(); @@ -938,10 +874,9 @@ mod tests { let tx = Transaction::new(&table); // Add "email" to the "person" struct. - let action = tx.update_schema().add_column_to( - "person", - "email", - Type::Primitive(PrimitiveType::String), + let action = tx.update_schema().add_column( + AddColumn::optional("email", Type::Primitive(PrimitiveType::String)) + .with_parent("person"), ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); @@ -970,11 +905,10 @@ mod tests { let table = make_v2_table_with_nested(); let tx = Transaction::new(&table); - let action = tx.update_schema().add_column_to_with_doc( - "person", - "phone", - Type::Primitive(PrimitiveType::String), - "Phone number", + let action = tx.update_schema().add_column( + AddColumn::optional("phone", Type::Primitive(PrimitiveType::String)) + .with_parent("person") + .with_doc("Phone number"), ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); @@ -999,10 +933,9 @@ mod tests { // "tags" is a list. Adding to the list navigates to its // element struct automatically. - let action = tx.update_schema().add_column_to( - "tags", - "score", - Type::Primitive(PrimitiveType::Double), + let action = tx.update_schema().add_column( + AddColumn::optional("score", Type::Primitive(PrimitiveType::Double)) + .with_parent("tags"), ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); @@ -1032,10 +965,9 @@ mod tests { // "props" is a map. Adding to the map navigates to its // value struct automatically. - let action = tx.update_schema().add_column_to( - "props", - "version", - Type::Primitive(PrimitiveType::Int), + let action = tx.update_schema().add_column( + AddColumn::optional("version", Type::Primitive(PrimitiveType::Int)) + .with_parent("props"), ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); @@ -1060,10 +992,9 @@ mod tests { let table = make_v2_table_with_nested(); let tx = Transaction::new(&table); - let action = tx.update_schema().add_column_to( - "nonexistent", - "col", - Type::Primitive(PrimitiveType::Int), + let action = tx.update_schema().add_column( + AddColumn::optional("col", Type::Primitive(PrimitiveType::Int)) + .with_parent("nonexistent"), ); let err = match Arc::new(action).commit(&table).await { @@ -1084,9 +1015,9 @@ mod tests { let tx = Transaction::new(&table); // "x" is a primitive (long), not a struct. - let action = - tx.update_schema() - .add_column_to("x", "col", Type::Primitive(PrimitiveType::Int)); + let action = tx.update_schema().add_column( + AddColumn::optional("col", Type::Primitive(PrimitiveType::Int)).with_parent("x"), + ); let err = match Arc::new(action).commit(&table).await { Err(e) => e, @@ -1106,10 +1037,9 @@ mod tests { let tx = Transaction::new(&table); // "name" already exists in the "person" struct. - let action = tx.update_schema().add_column_to( - "person", - "name", - Type::Primitive(PrimitiveType::String), + let action = tx.update_schema().add_column( + AddColumn::optional("name", Type::Primitive(PrimitiveType::String)) + .with_parent("person"), ); let err = match Arc::new(action).commit(&table).await { @@ -1132,8 +1062,14 @@ mod tests { // Add a root column and a nested column in the same action. let action = tx .update_schema() - .add_column("root_col", Type::Primitive(PrimitiveType::Boolean)) - .add_column_to("person", "email", Type::Primitive(PrimitiveType::String)); + .add_column(AddColumn::optional( + "root_col", + Type::Primitive(PrimitiveType::Boolean), + )) + .add_column( + AddColumn::optional("email", Type::Primitive(PrimitiveType::String)) + .with_parent("person"), + ); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); @@ -1158,19 +1094,20 @@ mod tests { #[tokio::test] async fn test_add_nested_struct_type_with_fresh_ids() { - // Exercises the assign_fresh_ids bug fix: adding a new column whose TYPE - // contains nested fields (e.g. a struct column). All sub-fields must receive - // fresh IDs, not placeholder 0. + // Adding a new column whose TYPE contains nested fields (e.g. a struct column). All sub-fields must receive + // fresh IDs, not placeholder `DEFAULT_ID`. let table = make_v2_table(); let tx = Transaction::new(&table); - let action = tx.update_schema().add_column( + let action = tx.update_schema().add_column(AddColumn::optional( "address", Type::Struct(StructType::new(vec![ - NestedField::optional(0, "street", Type::Primitive(PrimitiveType::String)).into(), - NestedField::optional(0, "city", Type::Primitive(PrimitiveType::String)).into(), + NestedField::optional(DEFAULT_ID, "street", Type::Primitive(PrimitiveType::String)) + .into(), + NestedField::optional(DEFAULT_ID, "city", Type::Primitive(PrimitiveType::String)) + .into(), ])), - ); + )); let mut action_commit = Arc::new(action).commit(&table).await.unwrap(); let updates = action_commit.take_updates(); diff --git a/crates/integration_tests/tests/update_schema.rs b/crates/integration_tests/tests/update_schema.rs index 17b106f669..b49a66cc7c 100644 --- a/crates/integration_tests/tests/update_schema.rs +++ b/crates/integration_tests/tests/update_schema.rs @@ -24,7 +24,7 @@ use std::sync::Arc; use arrow_array::{ArrayRef, BooleanArray, Int32Array, RecordBatch, StringArray, StructArray}; use common::{random_ns, test_schema}; use futures::TryStreamExt; -use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::transaction::{AddColumn, ApplyTransactionAction, Transaction}; use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; use iceberg::writer::file_writer::ParquetWriterBuilder; use iceberg::writer::file_writer::location_generator::{ @@ -146,10 +146,10 @@ async fn test_add_field() { // Add a new optional primitive field to the table let tx = Transaction::new(&table); - let add_action = tx.update_schema().add_column( + let add_action = tx.update_schema().add_column(AddColumn::optional( "a", iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::Int), - ); + )); let tx = add_action.apply(tx).unwrap(); let table = tx.commit(&rest_catalog).await.unwrap(); @@ -168,7 +168,7 @@ async fn test_add_field() { // Add a struct column, then add a nested column inside it let tx = Transaction::new(&table); - let add_action = tx.update_schema().add_column( + let add_action = tx.update_schema().add_column(AddColumn::optional( "info", iceberg::spec::Type::Struct(iceberg::spec::StructType::new(vec![Arc::new( iceberg::spec::NestedField::optional( @@ -177,7 +177,7 @@ async fn test_add_field() { iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), ), )])), - ); + )); let tx = add_action.apply(tx).unwrap(); let table = tx.commit(&rest_catalog).await.unwrap(); @@ -200,10 +200,12 @@ async fn test_add_field() { // Add a nested column to the struct let tx = Transaction::new(&table); - let add_action = tx.update_schema().add_column_to( - "info", - "zip", - iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), + let add_action = tx.update_schema().add_column( + AddColumn::optional( + "zip", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), + ) + .with_parent("info"), ); let tx = add_action.apply(tx).unwrap(); let table = tx.commit(&rest_catalog).await.unwrap(); From 2ad6598083083e50892cc14c5024a1d79a5a0242 Mon Sep 17 00:00:00 2001 From: Denis Semenov Date: Mon, 30 Mar 2026 17:32:43 +0300 Subject: [PATCH 09/10] Add id field to the builder --- crates/iceberg/src/transaction/update_schema.rs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index 5d0b435d8d..98e2476233 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -48,6 +48,8 @@ pub struct AddColumn { #[builder(default = false)] required: bool, field_type: Type, + #[builder(default = None)] + id: Option, #[builder(default = None, setter(strip_option, into))] doc: Option, #[builder(default = None, setter(strip_option))] @@ -91,7 +93,7 @@ impl AddColumn { fn to_nested_field(&self) -> NestedFieldRef { let mut field = NestedField::new( - DEFAULT_ID, + self.id.unwrap_or(DEFAULT_ID), self.name.clone(), self.field_type.clone(), self.required, @@ -457,7 +459,7 @@ impl TransactionAction for UpdateSchemaAction { }; // Assign fresh IDs immediately, preserving insertion order. - let field = if self.auto_assign_ids { + let field = if self.auto_assign_ids && pending_field.id == DEFAULT_ID { assign_fresh_ids(&pending_field, &mut last_column_id) } else { pending_field From e87889301b7913a19469501d5855bef551f971d8 Mon Sep 17 00:00:00 2001 From: Denis Semenov Date: Thu, 16 Apr 2026 12:21:44 +0300 Subject: [PATCH 10/10] Review fixes --- .../iceberg/src/transaction/update_schema.rs | 300 +++++++++++------- 1 file changed, 194 insertions(+), 106 deletions(-) diff --git a/crates/iceberg/src/transaction/update_schema.rs b/crates/iceberg/src/transaction/update_schema.rs index 98e2476233..ca1d352103 100644 --- a/crates/iceberg/src/transaction/update_schema.rs +++ b/crates/iceberg/src/transaction/update_schema.rs @@ -31,7 +31,7 @@ use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate}; /// Sentinel parent ID representing the table root (top-level columns). const TABLE_ROOT_ID: i32 = -1; // Default ID for a new column. This will be re-assigned to a fresh ID at commit time. -const DEFAULT_ID: i32 = 0; +const DEFAULT_FIELD_ID: i32 = 0; #[derive(TypedBuilder)] /// Declarative specification for adding a column in [`UpdateSchemaAction`]. @@ -60,18 +60,18 @@ pub struct AddColumn { impl AddColumn { /// Create a root-level optional column specification. - pub fn optional(name: impl ToString, field_type: Type) -> Self { + pub fn optional(name: impl Into, field_type: Type) -> Self { Self::builder() - .name(name.to_string()) + .name(name.into()) .field_type(field_type) .required(false) .build() } /// Create a root-level required column specification. - pub fn required(name: impl ToString, field_type: Type, initial_default: Literal) -> Self { + pub fn required(name: impl Into, field_type: Type, initial_default: Literal) -> Self { Self::builder() - .name(name.to_string()) + .name(name.into()) .field_type(field_type) .required(true) .initial_default(initial_default.clone()) @@ -80,20 +80,20 @@ impl AddColumn { } /// Return a copy with an updated parent path. - pub fn with_parent(mut self, parent: impl ToString) -> Self { - self.parent = Some(parent.to_string()); + pub fn with_parent(mut self, parent: impl Into) -> Self { + self.parent = Some(parent.into()); self } /// Return a copy with an updated doc string. - pub fn with_doc(mut self, doc: impl ToString) -> Self { - self.doc = Some(doc.to_string()); + pub fn with_doc(mut self, doc: impl Into) -> Self { + self.doc = Some(doc.into()); self } fn to_nested_field(&self) -> NestedFieldRef { let mut field = NestedField::new( - self.id.unwrap_or(DEFAULT_ID), + self.id.unwrap_or(DEFAULT_FIELD_ID), self.name.clone(), self.field_type.clone(), self.required, @@ -130,7 +130,7 @@ impl AddColumn { /// ``` pub struct UpdateSchemaAction { additions: Vec, - deletes: Vec, + deletions: Vec, auto_assign_ids: bool, } @@ -139,13 +139,11 @@ impl UpdateSchemaAction { pub(crate) fn new() -> Self { Self { additions: Vec::new(), - deletes: Vec::new(), + deletions: Vec::new(), auto_assign_ids: true, } } - // --- Root-level additions --- - /// Add a column to the table schema. /// /// To add a root-level column, leave `AddColumn::parent` as `None`. @@ -156,13 +154,11 @@ impl UpdateSchemaAction { self } - // --- Other builder methods --- - /// Record a column deletion by name. /// /// At commit time, the column must exist in the current schema. - pub fn delete_column(mut self, name: impl ToString) -> Self { - self.deletes.push(name.to_string()); + pub fn delete_column(mut self, name: impl Into) -> Self { + self.deletions.push(name.into()); self } @@ -172,8 +168,6 @@ impl UpdateSchemaAction { self.auto_assign_ids = false; self } - - // --- Internal helpers --- } // --------------------------------------------------------------------------- @@ -355,15 +349,115 @@ fn rebuild_field( // TransactionAction implementation // --------------------------------------------------------------------------- +/// Validate and process a single column addition. +/// +/// Returns the parent ID where the column should be added and the field with +/// assigned IDs (if auto-assignment is enabled). +fn process_addition( + add: &AddColumn, + base_schema: &Schema, + delete_ids: &HashSet, + auto_assign_ids: bool, + last_column_id: &mut i32, +) -> Result<(i32, NestedFieldRef)> { + let pending_field = add.to_nested_field(); + + // Check that name does not contain ".". + if pending_field.name.contains('.') { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column with ambiguous name: {}. Use `AddColumn::with_parent` to add a column to a nested struct.", + pending_field.name + ), + )); + } + + // Required columns without an initial default need allow_incompatible_changes. + if pending_field.required && pending_field.initial_default.is_none() { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Incompatible change: cannot add required column without an initial default: {}", + pending_field.name + ), + )); + } + + let parent_id = match &add.parent { + None => { + // Root-level: check name conflict against root-level fields. + if let Some(existing) = base_schema.field_by_name(&pending_field.name) + && !delete_ids.contains(&existing.id) + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column, name already exists: {}", + pending_field.name + ), + )); + } + TABLE_ROOT_ID + } + Some(parent_path) => { + // Nested: resolve parent, check name conflict within parent struct. + let (parent_field_id, parent_struct) = resolve_parent_target(base_schema, parent_path)?; + + // Check if the parent field is being deleted - cannot add to a deleted parent. + let parent_field = base_schema + .field_by_name(parent_path) + .expect("parent field should exist (already validated by resolve_parent_target)"); + + if delete_ids.contains(&parent_field.id) { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column to parent that is being deleted: '{}'", + parent_path + ), + )); + } + + // Check for name conflict within the parent struct. + if let Some(existing) = parent_struct.field_by_name(&pending_field.name) + && !delete_ids.contains(&existing.id) + { + return Err(Error::new( + ErrorKind::PreconditionFailed, + format!( + "Cannot add column, name already exists in '{}': {}", + parent_path, pending_field.name + ), + )); + } + + parent_field_id + } + }; + + // Assign fresh IDs if auto-assignment is enabled and field has placeholder ID. + let field = if auto_assign_ids && pending_field.id == DEFAULT_FIELD_ID { + assign_fresh_ids(&pending_field, last_column_id) + } else { + pending_field + }; + + Ok((parent_id, field)) +} + #[async_trait] impl TransactionAction for UpdateSchemaAction { async fn commit(self: Arc, table: &Table) -> Result { let base_schema = table.metadata().current_schema(); let mut last_column_id = table.metadata().last_column_id(); + // Collect identifier field IDs for checking during deletion. + let identifier_field_ids: HashSet = base_schema.identifier_field_ids().collect(); + // --- 1. Validate deletes --- let delete_ids = self - .deletes + .deletions .iter() .map(|name: &String| { base_schema @@ -375,103 +469,38 @@ impl TransactionAction for UpdateSchemaAction { ) }) .and_then(|field| { - match base_schema - .identifier_field_ids() - .find(|id| *id == field.id) - { - Some(_) => Err(Error::new( + if identifier_field_ids.contains(&field.id) { + Err(Error::new( ErrorKind::PreconditionFailed, format!("Cannot delete identifier field: {name}"), - )), - None => Ok(field.id), + )) + } else { + Ok(field.id) } }) }) .collect::>>()?; - // --- 2. Resolve parents, validate additions, assign IDs, and group by parent ID --- + // --- 2. Process additions and group by parent ID --- // We assign IDs inline (before grouping) to preserve the caller's insertion order, // since HashMap iteration order is non-deterministic. let mut additions_by_parent: HashMap> = HashMap::new(); for add in &self.additions { - let pending_field = add.to_nested_field(); - - // Check that name does not contain ".". - if pending_field.name.contains('.') { - return Err(Error::new( - ErrorKind::PreconditionFailed, - format!( - "Cannot add column with ambiguous name: {}. Use `AddColumn::with_parent` to add a column to a nested struct.", - pending_field.name - ), - )); - } - - // Required columns without an initial default need allow_incompatible_changes. - if pending_field.required && pending_field.initial_default.is_none() { - return Err(Error::new( - ErrorKind::PreconditionFailed, - format!( - "Incompatible change: cannot add required column without an initial default: {}", - pending_field.name - ), - )); - } - - let parent_id = match &add.parent { - None => { - // Root-level: check name conflict against root-level fields. - if let Some(existing) = base_schema.field_by_name(&pending_field.name) - && !delete_ids.contains(&existing.id) - { - return Err(Error::new( - ErrorKind::PreconditionFailed, - format!( - "Cannot add column, name already exists: {}", - pending_field.name - ), - )); - } - TABLE_ROOT_ID - } - Some(parent_path) => { - // Nested: resolve parent, check name conflict within parent struct. - let (parent_id, parent_struct) = - resolve_parent_target(base_schema, parent_path)?; - - if parent_struct.fields().iter().any(|f| { - f.name == pending_field.name - && !delete_ids.contains(&f.id) - && !delete_ids.contains(&parent_id) - }) { - return Err(Error::new( - ErrorKind::PreconditionFailed, - format!( - "Cannot add column, name already exists in '{}': {}", - parent_path, pending_field.name - ), - )); - } - - parent_id - } - }; - - // Assign fresh IDs immediately, preserving insertion order. - let field = if self.auto_assign_ids && pending_field.id == DEFAULT_ID { - assign_fresh_ids(&pending_field, &mut last_column_id) - } else { - pending_field - }; - + let (parent_id, field) = process_addition( + add, + base_schema, + &delete_ids, + self.auto_assign_ids, + &mut last_column_id, + )?; additions_by_parent .entry(parent_id) .or_default() .push(field); } - // --- 4. Rebuild the schema tree with additions and deletions --- + // --- 3. Rebuild the schema tree with additions and deletions --- let new_fields = rebuild_fields( base_schema.as_struct().fields(), &additions_by_parent, @@ -479,7 +508,7 @@ impl TransactionAction for UpdateSchemaAction { TABLE_ROOT_ID, ); - // --- 5. Build the new schema --- + // --- 4. Build the new schema --- let schema = Schema::builder() .with_fields(new_fields) .with_identifier_field_ids(base_schema.identifier_field_ids()) @@ -510,7 +539,7 @@ mod tests { use crate::transaction::Transaction; use crate::transaction::action::{ApplyTransactionAction, TransactionAction}; use crate::transaction::tests::make_v2_table; - use crate::transaction::update_schema::{AddColumn, DEFAULT_ID, UpdateSchemaAction}; + use crate::transaction::update_schema::{AddColumn, DEFAULT_FIELD_ID, UpdateSchemaAction}; use crate::{ErrorKind, TableIdent, TableRequirement, TableUpdate}; // The V2 test table has: @@ -1104,10 +1133,18 @@ mod tests { let action = tx.update_schema().add_column(AddColumn::optional( "address", Type::Struct(StructType::new(vec![ - NestedField::optional(DEFAULT_ID, "street", Type::Primitive(PrimitiveType::String)) - .into(), - NestedField::optional(DEFAULT_ID, "city", Type::Primitive(PrimitiveType::String)) - .into(), + NestedField::optional( + DEFAULT_FIELD_ID, + "street", + Type::Primitive(PrimitiveType::String), + ) + .into(), + NestedField::optional( + DEFAULT_FIELD_ID, + "city", + Type::Primitive(PrimitiveType::String), + ) + .into(), ])), )); @@ -1136,4 +1173,55 @@ mod tests { .expect("address.city should exist"); assert_eq!(city.id, 6); } + + // ----------------------------------------------------------------------- + // Delete parent and add to deleted parent tests + // ----------------------------------------------------------------------- + + #[tokio::test] + async fn test_add_column_to_deleted_column_parent_fails() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // Delete "person" struct and try to add a column to it - should fail. + let action = tx.update_schema().delete_column("person").add_column( + AddColumn::optional("email", Type::Primitive(PrimitiveType::String)) + .with_parent("person"), + ); + + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("should reject adding to a deleted parent"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("being deleted"), + "error should mention deleted parent, got: {}", + err.message() + ); + } + + #[tokio::test] + async fn test_add_column_with_same_name_as_child_of_deleted_parent_fails() { + let table = make_v2_table_with_nested(); + let tx = Transaction::new(&table); + + // Delete "person" struct (which has a child "name") and try to add a column + // named "name" to the deleted parent - should fail because parent is being deleted. + let action = tx.update_schema().delete_column("person").add_column( + AddColumn::optional("name", Type::Primitive(PrimitiveType::String)) + .with_parent("person"), + ); + + let err = match Arc::new(action).commit(&table).await { + Err(e) => e, + Ok(_) => panic!("should reject adding to a deleted parent"), + }; + assert_eq!(err.kind(), ErrorKind::PreconditionFailed); + assert!( + err.message().contains("being deleted"), + "error should mention deleted parent, got: {}", + err.message() + ); + } }