iceberg/transaction/
update_schema.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::{HashMap, HashSet};
19use std::sync::Arc;
20
21use async_trait::async_trait;
22use typed_builder::TypedBuilder;
23
24use crate::spec::{
25    ListType, Literal, MapType, NestedField, NestedFieldRef, SCHEMA_NAME_DELIMITER, Schema,
26    StructType, Type,
27};
28use crate::table::Table;
29use crate::transaction::action::{ActionCommit, TransactionAction};
30use crate::{Error, ErrorKind, Result, TableRequirement, TableUpdate};
31
32// Default ID for a new column. This will be re-assigned to a fresh ID at commit time.
33const DEFAULT_FIELD_ID: i32 = 0;
34
35/// Declarative specification for adding a column in an [`UpdateSchemaAction`].
36///
37/// Use helper constructors such as [`AddColumn::optional`] and [`AddColumn::required`],
38/// optionally combined with the builder's `parent` and `doc` setters via
39/// [`AddColumn::builder`], then pass the value to [`UpdateSchemaAction::add_column`].
40#[derive(TypedBuilder)]
41pub struct AddColumn {
42    #[builder(default = None, setter(strip_option, into))]
43    parent: Option<String>,
44    #[builder(setter(into))]
45    name: String,
46    #[builder(default = false)]
47    required: bool,
48    field_type: Type,
49    #[builder(default = None, setter(strip_option, into))]
50    doc: Option<String>,
51    #[builder(default = None, setter(strip_option))]
52    initial_default: Option<Literal>,
53    #[builder(default = None, setter(strip_option))]
54    write_default: Option<Literal>,
55}
56
57impl AddColumn {
58    /// Create a root-level optional column specification.
59    pub fn optional(name: impl ToString, field_type: Type) -> Self {
60        Self::builder()
61            .name(name.to_string())
62            .field_type(field_type)
63            .required(false)
64            .build()
65    }
66
67    /// Create a root-level required column specification.
68    pub fn required(name: impl ToString, field_type: Type, initial_default: Literal) -> Self {
69        Self::builder()
70            .name(name.to_string())
71            .field_type(field_type)
72            .required(true)
73            .initial_default(initial_default.clone())
74            .write_default(initial_default)
75            .build()
76    }
77
78    fn to_nested_field(&self) -> NestedFieldRef {
79        let mut field = NestedField::new(
80            DEFAULT_FIELD_ID,
81            self.name.clone(),
82            self.field_type.clone(),
83            self.required,
84        );
85
86        field.doc = self.doc.clone();
87        field.initial_default = self.initial_default.clone();
88        field.write_default = self.write_default.clone();
89        Arc::new(field)
90    }
91}
92
93/// Schema evolution API modeled after the Java `SchemaUpdate` implementation.
94///
95/// This action accumulates schema modifications (column additions and deletions)
96/// via builder methods. At commit time, it validates all operations against the
97/// current table schema, auto-assigns field IDs from `table.metadata().last_column_id()`,
98/// builds a new schema, and emits `AddSchema` + `SetCurrentSchema` updates with a
99/// `CurrentSchemaIdMatch` requirement.
100///
101/// # Example
102///
103/// ```ignore
104/// let tx = Transaction::new(&table);
105/// let action = tx.update_schema()
106///     .add_column(AddColumn::optional("new_col", Type::Primitive(PrimitiveType::Int)))
107///     .add_column(
108///         AddColumn::optional("email", Type::Primitive(PrimitiveType::String))
109///             .with_parent("person")
110///     )
111///     .delete_column("old_col");
112/// let tx = action.apply(tx).unwrap();
113/// let table = tx.commit(&catalog).await.unwrap();
114/// ```
115pub struct UpdateSchemaAction {
116    additions: Vec<AddColumn>,
117    deletes: Vec<String>,
118}
119
120impl UpdateSchemaAction {
121    /// Creates a new empty `UpdateSchemaAction`.
122    pub(crate) fn new() -> Self {
123        Self {
124            additions: Vec::new(),
125            deletes: Vec::new(),
126        }
127    }
128
129    // --- Root-level additions ---
130
131    /// Add a column to the table schema.
132    ///
133    /// To add a root-level column, leave `AddColumn::parent` as `None`.
134    /// For nested additions, set a parent path.
135    /// If the parent resolves to a map/list, the column is added to map value/list element.
136    pub fn add_column(mut self, add_column: AddColumn) -> Self {
137        self.additions.push(add_column);
138        self
139    }
140
141    // --- Other builder methods ---
142
143    /// Record a column deletion by name.
144    ///
145    /// At commit time, the column must exist in the current schema.
146    pub fn delete_column(mut self, name: impl ToString) -> Self {
147        self.deletes.push(name.to_string());
148        self
149    }
150}
151
152// ---------------------------------------------------------------------------
153// ID assignment helpers
154// ---------------------------------------------------------------------------
155
156/// Recursively assign fresh field IDs to a `NestedField` and all its nested sub-fields.
157///
158/// This follows the same recursive pattern as `ReassignFieldIds::reassign_ids_visit_type`
159/// from `crate::spec::schema::id_reassigner`, but operates on new fields with placeholder
160/// IDs rather than reassigning an existing schema. `ReassignFieldIds` cannot be used
161/// directly here because it rejects duplicate old IDs (all new fields share placeholder
162/// ID `DEFAULT_FIELD_ID`).
163fn assign_fresh_ids(field: &NestedField, next_id: &mut i32) -> NestedFieldRef {
164    *next_id += 1;
165    let new_id = *next_id;
166    let new_type = assign_fresh_ids_to_type(&field.field_type, next_id);
167
168    Arc::new(NestedField {
169        id: new_id,
170        name: field.name.clone(),
171        required: field.required,
172        field_type: Box::new(new_type),
173        doc: field.doc.clone(),
174        initial_default: field.initial_default.clone(),
175        write_default: field.write_default.clone(),
176    })
177}
178
179/// Recursively assign fresh field IDs to all nested fields within a `Type`.
180fn assign_fresh_ids_to_type(field_type: &Type, next_id: &mut i32) -> Type {
181    match field_type {
182        Type::Primitive(_) => field_type.clone(),
183        // Variant carries no nested fields, so there is nothing to reassign
184        // (matches id_reassigner.rs).
185        Type::Variant(v) => Type::Variant(*v),
186        Type::Struct(struct_type) => {
187            let new_fields: Vec<NestedFieldRef> = struct_type
188                .fields()
189                .iter()
190                .map(|f| assign_fresh_ids(f, next_id))
191                .collect();
192            Type::Struct(StructType::new(new_fields))
193        }
194        Type::List(list_type) => {
195            let new_element = assign_fresh_ids(&list_type.element_field, next_id);
196            Type::List(ListType {
197                element_field: new_element,
198            })
199        }
200        Type::Map(map_type) => {
201            let new_key = assign_fresh_ids(&map_type.key_field, next_id);
202            let new_value = assign_fresh_ids(&map_type.value_field, next_id);
203            Type::Map(MapType {
204                key_field: new_key,
205                value_field: new_value,
206            })
207        }
208    }
209}
210
211// ---------------------------------------------------------------------------
212// Parent path resolution
213// ---------------------------------------------------------------------------
214
215/// Resolve a parent path to the target struct's parent field ID and a reference
216/// to its `StructType`.
217///
218/// If the parent is a map, navigates to the value field. If a list, navigates to
219/// the element field. The final target must be a struct type.
220fn resolve_parent_target<'a>(
221    base_schema: &'a Schema,
222    parent: &str,
223) -> Result<(i32, &'a StructType)> {
224    base_schema
225        .field_by_name(parent)
226        .ok_or_else(|| {
227            Error::new(
228                ErrorKind::PreconditionFailed,
229                format!("Cannot add column: parent '{parent}' not found"),
230            )
231        })
232        .and_then(|parent_field| match parent_field.field_type.as_ref() {
233            Type::Struct(s) => Ok((parent_field.id, s)),
234            Type::Map(m) => match m.value_field.field_type.as_ref() {
235                Type::Struct(s) => Ok((m.value_field.id, s)),
236                _ => Err(Error::new(
237                    ErrorKind::PreconditionFailed,
238                    format!("Cannot add column: map value of '{parent}' is not a struct"),
239                )),
240            },
241            Type::List(l) => match l.element_field.field_type.as_ref() {
242                Type::Struct(s) => Ok((l.element_field.id, s)),
243                _ => Err(Error::new(
244                    ErrorKind::PreconditionFailed,
245                    format!("Cannot add column: list element of '{parent}' is not a struct"),
246                )),
247            },
248            _ => Err(Error::new(
249                ErrorKind::PreconditionFailed,
250                format!("Cannot add column: parent '{parent}' is not a struct, map, or list"),
251            )),
252        })
253}
254
255// ---------------------------------------------------------------------------
256// Schema tree rebuild
257// ---------------------------------------------------------------------------
258
259/// Rebuild a slice of fields, applying deletions and additions at every level,
260/// plus any additions keyed by `parent_id` (`None` represents the table root).
261fn rebuild_fields(
262    fields: &[NestedFieldRef],
263    adds: &HashMap<Option<i32>, Vec<NestedFieldRef>>,
264    delete_ids: &HashSet<i32>,
265    parent_id: Option<i32>,
266) -> Vec<NestedFieldRef> {
267    fields
268        .iter()
269        .filter(|f| !delete_ids.contains(&f.id))
270        .map(|f| rebuild_field(f, adds, delete_ids))
271        .chain(adds.get(&parent_id).into_iter().flatten().cloned())
272        .collect()
273}
274
275/// Recursively rebuild a single field. If the field (or any descendant) is a struct
276/// that has pending additions, those additions are appended to the struct's fields.
277/// Fields whose IDs appear in `delete_ids` are filtered out at every struct level.
278fn rebuild_field(
279    field: &NestedFieldRef,
280    adds: &HashMap<Option<i32>, Vec<NestedFieldRef>>,
281    delete_ids: &HashSet<i32>,
282) -> NestedFieldRef {
283    match field.field_type.as_ref() {
284        Type::Primitive(_) | Type::Variant(_) => field.clone(),
285        Type::Struct(s) => {
286            let new_fields = rebuild_fields(s.fields(), adds, delete_ids, Some(field.id));
287            Arc::new(NestedField {
288                id: field.id,
289                name: field.name.clone(),
290                required: field.required,
291                field_type: Box::new(Type::Struct(StructType::new(new_fields))),
292                doc: field.doc.clone(),
293                initial_default: field.initial_default.clone(),
294                write_default: field.write_default.clone(),
295            })
296        }
297        Type::List(l) => {
298            let new_element = rebuild_field(&l.element_field, adds, delete_ids);
299            Arc::new(NestedField {
300                id: field.id,
301                name: field.name.clone(),
302                required: field.required,
303                field_type: Box::new(Type::List(ListType {
304                    element_field: new_element,
305                })),
306                doc: field.doc.clone(),
307                initial_default: field.initial_default.clone(),
308                write_default: field.write_default.clone(),
309            })
310        }
311        Type::Map(m) => {
312            let new_key = rebuild_field(&m.key_field, adds, delete_ids);
313            let new_value = rebuild_field(&m.value_field, adds, delete_ids);
314            Arc::new(NestedField {
315                id: field.id,
316                name: field.name.clone(),
317                required: field.required,
318                field_type: Box::new(Type::Map(MapType {
319                    key_field: new_key,
320                    value_field: new_value,
321                })),
322                doc: field.doc.clone(),
323                initial_default: field.initial_default.clone(),
324                write_default: field.write_default.clone(),
325            })
326        }
327    }
328}
329
330// ---------------------------------------------------------------------------
331// TransactionAction implementation
332// ---------------------------------------------------------------------------
333
334#[async_trait]
335impl TransactionAction for UpdateSchemaAction {
336    async fn commit(self: Arc<Self>, table: &Table) -> Result<ActionCommit> {
337        let base_schema = table.metadata().current_schema();
338        let mut last_column_id = table.metadata().last_column_id();
339
340        // --- 1. Validate deletes ---
341        let delete_ids = self
342            .deletes
343            .iter()
344            .map(|name: &String| {
345                base_schema
346                    .field_by_name(name)
347                    .ok_or_else(|| {
348                        Error::new(
349                            ErrorKind::PreconditionFailed,
350                            format!("Cannot delete missing column: {name}"),
351                        )
352                    })
353                    .and_then(|field| {
354                        match base_schema
355                            .identifier_field_ids()
356                            .find(|id| *id == field.id)
357                        {
358                            Some(_) => Err(Error::new(
359                                ErrorKind::PreconditionFailed,
360                                format!("Cannot delete identifier field: {name}"),
361                            )),
362                            None => Ok(field.id),
363                        }
364                    })
365            })
366            .collect::<Result<HashSet<i32>>>()?;
367
368        // --- 2. Resolve parents, validate additions, assign IDs, and group by parent ID ---
369        // We assign IDs inline (before grouping) to preserve the caller's insertion order,
370        // since HashMap iteration order is non-deterministic.
371        let mut additions_by_parent: HashMap<Option<i32>, Vec<NestedFieldRef>> = HashMap::new();
372
373        for add in &self.additions {
374            let pending_field = add.to_nested_field();
375
376            // Check that name does not contain `SCHEMA_NAME_DELIMITER`.
377            if pending_field.name.contains(SCHEMA_NAME_DELIMITER) {
378                return Err(Error::new(
379                    ErrorKind::PreconditionFailed,
380                    format!(
381                        "Cannot add column with ambiguous name: {}. Use `AddColumn::with_parent` to add a column to a nested struct.",
382                        pending_field.name
383                    ),
384                ));
385            }
386
387            // Required columns without an initial default need allow_incompatible_changes.
388            if pending_field.required && pending_field.initial_default.is_none() {
389                return Err(Error::new(
390                    ErrorKind::PreconditionFailed,
391                    format!(
392                        "Incompatible change: cannot add required column without an initial default: {}",
393                        pending_field.name
394                    ),
395                ));
396            }
397
398            let parent_id = match &add.parent {
399                None => {
400                    // Root-level: check name conflict against root-level fields.
401                    if let Some(existing) = base_schema.field_by_name(&pending_field.name)
402                        && !delete_ids.contains(&existing.id)
403                    {
404                        return Err(Error::new(
405                            ErrorKind::PreconditionFailed,
406                            format!(
407                                "Cannot add column, name already exists: {}",
408                                pending_field.name
409                            ),
410                        ));
411                    }
412                    None
413                }
414                Some(parent_path) => {
415                    // Nested: resolve parent, check name conflict within parent struct.
416                    let (resolved_parent_id, parent_struct) =
417                        resolve_parent_target(base_schema, parent_path)?;
418
419                    if parent_struct.fields().iter().any(|f| {
420                        f.name == pending_field.name
421                            && !delete_ids.contains(&f.id)
422                            && !delete_ids.contains(&resolved_parent_id)
423                    }) {
424                        return Err(Error::new(
425                            ErrorKind::PreconditionFailed,
426                            format!(
427                                "Cannot add column, name already exists in '{}': {}",
428                                parent_path, pending_field.name
429                            ),
430                        ));
431                    }
432
433                    Some(resolved_parent_id)
434                }
435            };
436
437            // Assign fresh IDs immediately, preserving insertion order.
438            let field = assign_fresh_ids(&pending_field, &mut last_column_id);
439
440            additions_by_parent
441                .entry(parent_id)
442                .or_default()
443                .push(field);
444        }
445
446        // --- 4. Rebuild the schema tree with additions and deletions ---
447        let new_fields = rebuild_fields(
448            base_schema.as_struct().fields(),
449            &additions_by_parent,
450            &delete_ids,
451            None,
452        );
453
454        // --- 5. Build the new schema ---
455        let schema = Schema::builder()
456            .with_fields(new_fields)
457            .with_identifier_field_ids(base_schema.identifier_field_ids())
458            .build()?;
459
460        let updates = vec![
461            TableUpdate::AddSchema { schema },
462            TableUpdate::SetCurrentSchema { schema_id: -1 },
463        ];
464
465        let requirements = vec![TableRequirement::CurrentSchemaIdMatch {
466            current_schema_id: base_schema.schema_id(),
467        }];
468
469        Ok(ActionCommit::new(updates, requirements))
470    }
471}
472
473#[cfg(test)]
474mod tests {
475    use std::io::BufReader;
476    use std::sync::Arc;
477
478    use as_any::Downcast;
479
480    use crate::spec::{
481        DEFAULT_SCHEMA_ID, Literal, NestedField, PrimitiveType, StructType, TableMetadata, Type,
482        VariantType,
483    };
484    use crate::table::Table;
485    use crate::transaction::Transaction;
486    use crate::transaction::action::{ApplyTransactionAction, TransactionAction};
487    use crate::transaction::tests::make_v2_table;
488    use crate::transaction::update_schema::{AddColumn, DEFAULT_FIELD_ID, UpdateSchemaAction};
489    use crate::{ErrorKind, TableIdent, TableRequirement, TableUpdate};
490
491    // The V2 test table has:
492    //   last_column_id: 3
493    //   current schema (id=1): x(1, req, long), y(2, req, long), z(3, req, long)
494    //   identifier_field_ids: [1, 2]
495
496    /// Build a V2 test table that includes nested types:
497    ///
498    ///   last_column_id: 14
499    ///   current schema (id=0):
500    ///     x(1, req, long)           -- identifier
501    ///     y(2, req, long)           -- identifier
502    ///     z(3, req, long)
503    ///     person(4, opt, struct)
504    ///       name(5, opt, string)
505    ///       age(6, req, int)
506    ///     tags(7, opt, list<struct>)
507    ///       element(8, req, struct)
508    ///         key(9, opt, string)
509    ///         value(10, opt, string)
510    ///     props(11, opt, map<string, struct>)
511    ///       key(12, req, string)
512    ///       value(13, req, struct)
513    ///         data(14, opt, string)
514    fn make_v2_table_with_nested() -> Table {
515        let json = r#"{
516            "format-version": 2,
517            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c2",
518            "location": "s3://bucket/test/location",
519            "last-sequence-number": 0,
520            "last-updated-ms": 1602638573590,
521            "last-column-id": 14,
522            "current-schema-id": 0,
523            "schemas": [
524                {
525                    "type": "struct",
526                    "schema-id": 0,
527                    "identifier-field-ids": [1, 2],
528                    "fields": [
529                        {"id": 1, "name": "x", "required": true, "type": "long"},
530                        {"id": 2, "name": "y", "required": true, "type": "long"},
531                        {"id": 3, "name": "z", "required": true, "type": "long"},
532                        {"id": 4, "name": "person", "required": false, "type": {
533                            "type": "struct",
534                            "fields": [
535                                {"id": 5, "name": "name", "required": false, "type": "string"},
536                                {"id": 6, "name": "age", "required": true, "type": "int"}
537                            ]
538                        }},
539                        {"id": 7, "name": "tags", "required": false, "type": {
540                            "type": "list",
541                            "element-id": 8,
542                            "element": {
543                                "type": "struct",
544                                "fields": [
545                                    {"id": 9, "name": "key", "required": false, "type": "string"},
546                                    {"id": 10, "name": "value", "required": false, "type": "string"}
547                                ]
548                            },
549                            "element-required": true
550                        }},
551                        {"id": 11, "name": "props", "required": false, "type": {
552                            "type": "map",
553                            "key-id": 12,
554                            "key": "string",
555                            "value-id": 13,
556                            "value": {
557                                "type": "struct",
558                                "fields": [
559                                    {"id": 14, "name": "data", "required": false, "type": "string"}
560                                ]
561                            },
562                            "value-required": true
563                        }}
564                    ]
565                }
566            ],
567            "default-spec-id": 0,
568            "partition-specs": [
569                {"spec-id": 0, "fields": []}
570            ],
571            "last-partition-id": 999,
572            "default-sort-order-id": 0,
573            "sort-orders": [
574                {"order-id": 0, "fields": []}
575            ],
576            "properties": {},
577            "current-snapshot-id": -1,
578            "snapshots": []
579        }"#;
580
581        let reader = BufReader::new(json.as_bytes());
582        let metadata = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
583
584        Table::builder()
585            .metadata(metadata)
586            .metadata_location("s3://bucket/test/location/metadata/v1.json".to_string())
587            .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
588            .file_io(crate::io::FileIO::new_with_memory())
589            .runtime(crate::test_utils::test_runtime())
590            .build()
591            .unwrap()
592    }
593
594    // -----------------------------------------------------------------------
595    // Existing root-level tests
596    // -----------------------------------------------------------------------
597
598    #[test]
599    fn test_assign_fresh_ids_variant() {
600        // Variant carries no sub-fields, so fresh-id assignment only renames the field
601        // itself and leaves the type untouched.
602        let mut next_id = 10;
603        let field = NestedField::optional(1, "data", Type::Variant(VariantType));
604        let assigned = super::assign_fresh_ids(&field, &mut next_id);
605
606        assert_eq!(assigned.id, 11);
607        assert_eq!(*assigned.field_type, Type::Variant(VariantType));
608        assert_eq!(next_id, 11);
609    }
610
611    #[tokio::test]
612    async fn test_add_column() {
613        let table = make_v2_table();
614        let tx = Transaction::new(&table);
615
616        let action = tx.update_schema().add_column(AddColumn::optional(
617            "new_col",
618            Type::Primitive(PrimitiveType::Int),
619        ));
620
621        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
622        let updates = action_commit.take_updates();
623        let requirements = action_commit.take_requirements();
624
625        assert_eq!(updates.len(), 2);
626
627        // Extract the new schema from the AddSchema update.
628        let new_schema = match &updates[0] {
629            TableUpdate::AddSchema { schema } => schema,
630            other => panic!("expected AddSchema, got {other:?}"),
631        };
632
633        let expected_schema = table
634            .metadata()
635            .current_schema()
636            .as_ref()
637            .clone()
638            .into_builder()
639            .with_schema_id(DEFAULT_SCHEMA_ID)
640            .with_fields([
641                NestedField::optional(4, "new_col", Type::Primitive(PrimitiveType::Int)).into(),
642            ])
643            .build()
644            .unwrap();
645        assert_eq!(new_schema, &expected_schema);
646
647        assert_eq!(updates[1], TableUpdate::SetCurrentSchema { schema_id: -1 });
648
649        // Verify requirement.
650        assert_eq!(requirements.len(), 1);
651        assert_eq!(requirements[0], TableRequirement::CurrentSchemaIdMatch {
652            current_schema_id: table.metadata().current_schema().schema_id()
653        });
654    }
655
656    #[tokio::test]
657    async fn test_add_column_with_doc() {
658        let table = make_v2_table();
659        let tx = Transaction::new(&table);
660
661        let action = tx.update_schema().add_column(
662            AddColumn::builder()
663                .name("documented_col")
664                .field_type(Type::Primitive(PrimitiveType::String))
665                .doc("A documented column")
666                .build(),
667        );
668
669        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
670        let updates = action_commit.take_updates();
671
672        let new_schema = match &updates[0] {
673            TableUpdate::AddSchema { schema } => schema,
674            other => panic!("expected AddSchema, got {other:?}"),
675        };
676
677        let field = new_schema
678            .field_by_name("documented_col")
679            .expect("documented_col should exist");
680        assert_eq!(field.id, 4);
681        assert!(!field.required);
682        assert_eq!(field.doc.as_deref(), Some("A documented column"));
683    }
684
685    #[tokio::test]
686    async fn test_add_required_column_with_initial_default() {
687        let table = make_v2_table();
688        let tx = Transaction::new(&table);
689
690        let action = tx.update_schema().add_column(AddColumn::required(
691            "req_col",
692            Type::Primitive(PrimitiveType::Int),
693            Literal::int(0),
694        ));
695
696        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
697        let updates = action_commit.take_updates();
698
699        let new_schema = match &updates[0] {
700            TableUpdate::AddSchema { schema } => schema,
701            other => panic!("expected AddSchema, got {other:?}"),
702        };
703
704        let field = new_schema
705            .field_by_name("req_col")
706            .expect("req_col should exist");
707        assert_eq!(field.id, 4);
708        assert!(field.required);
709        assert_eq!(field.initial_default, Some(Literal::int(0)));
710        assert_eq!(field.write_default, Some(Literal::int(0)));
711    }
712
713    #[tokio::test]
714    async fn test_add_column_name_conflict_fails() {
715        let table = make_v2_table();
716        let tx = Transaction::new(&table);
717
718        // "x" already exists in the V2 test schema.
719        let action = tx.update_schema().add_column(AddColumn::optional(
720            "x",
721            Type::Primitive(PrimitiveType::Int),
722        ));
723
724        let result = Arc::new(action).commit(&table).await;
725        let err = match result {
726            Err(e) => e,
727            Ok(_) => panic!("should reject adding a column with an existing name"),
728        };
729        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
730        assert!(
731            err.message().contains("already exists"),
732            "error should mention name conflict, got: {}",
733            err.message()
734        );
735    }
736
737    #[tokio::test]
738    async fn test_delete_column() {
739        let table = make_v2_table();
740        let tx = Transaction::new(&table);
741
742        // z is not an identifier field, so we can delete it.
743        let action = tx.update_schema().delete_column("z");
744
745        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
746        let updates = action_commit.take_updates();
747
748        let new_schema = match &updates[0] {
749            TableUpdate::AddSchema { schema } => schema,
750            other => panic!("expected AddSchema, got {other:?}"),
751        };
752
753        assert!(
754            new_schema.field_by_name("z").is_none(),
755            "z should be deleted"
756        );
757        assert!(new_schema.field_by_name("x").is_some());
758        assert!(new_schema.field_by_name("y").is_some());
759    }
760
761    #[tokio::test]
762    async fn test_delete_missing_column_fails() {
763        let table = make_v2_table();
764        let tx = Transaction::new(&table);
765
766        let action = tx.update_schema().delete_column("nonexistent");
767
768        let result = Arc::new(action).commit(&table).await;
769        let err = match result {
770            Err(e) => e,
771            Ok(_) => panic!("should reject deleting a non-existent column"),
772        };
773        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
774        assert!(
775            err.message().contains("nonexistent"),
776            "error should mention the missing column, got: {}",
777            err.message()
778        );
779    }
780
781    #[tokio::test]
782    async fn test_add_and_delete_combined() {
783        let table = make_v2_table();
784        let tx = Transaction::new(&table);
785
786        // Delete z, add a new column.
787        let action = tx
788            .update_schema()
789            .delete_column("z")
790            .add_column(AddColumn::optional(
791                "w",
792                Type::Primitive(PrimitiveType::Boolean),
793            ));
794
795        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
796        let updates = action_commit.take_updates();
797
798        let new_schema = match &updates[0] {
799            TableUpdate::AddSchema { schema } => schema,
800            other => panic!("expected AddSchema, got {other:?}"),
801        };
802
803        assert!(
804            new_schema.field_by_name("z").is_none(),
805            "z should be deleted"
806        );
807        let w = new_schema.field_by_name("w").expect("w should exist");
808        assert_eq!(w.id, 4);
809        assert!(!w.required);
810    }
811
812    #[tokio::test]
813    async fn test_delete_and_readd_same_name() {
814        let table = make_v2_table();
815        let tx = Transaction::new(&table);
816
817        // Delete z, then add a new column named z -- should succeed.
818        let action = tx
819            .update_schema()
820            .delete_column("z")
821            .add_column(AddColumn::optional(
822                "z",
823                Type::Primitive(PrimitiveType::Boolean),
824            ));
825
826        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
827        let updates = action_commit.take_updates();
828
829        let new_schema = match &updates[0] {
830            TableUpdate::AddSchema { schema } => schema,
831            other => panic!("expected AddSchema, got {other:?}"),
832        };
833
834        let z = new_schema
835            .field_by_name("z")
836            .expect("z should exist with new type");
837        assert_eq!(z.id, 4); // new ID, not the old 3
838        assert_eq!(*z.field_type, Type::Primitive(PrimitiveType::Boolean));
839    }
840
841    #[test]
842    fn test_apply() {
843        let table = make_v2_table();
844        let tx = Transaction::new(&table);
845
846        let tx = tx
847            .update_schema()
848            .add_column(AddColumn::optional(
849                "new_col",
850                Type::Primitive(PrimitiveType::Int),
851            ))
852            .apply(tx)
853            .unwrap();
854
855        assert_eq!(tx.actions.len(), 1);
856        (*tx.actions[0])
857            .downcast_ref::<UpdateSchemaAction>()
858            .expect("UpdateSchemaAction was not applied to Transaction!");
859    }
860
861    // -----------------------------------------------------------------------
862    // Nested add tests
863    // -----------------------------------------------------------------------
864
865    #[tokio::test]
866    async fn test_add_column_to_struct() {
867        let table = make_v2_table_with_nested();
868        let tx = Transaction::new(&table);
869
870        // Add "email" to the "person" struct.
871        let action = tx.update_schema().add_column(
872            AddColumn::builder()
873                .name("email")
874                .field_type(Type::Primitive(PrimitiveType::String))
875                .parent("person")
876                .build(),
877        );
878
879        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
880        let updates = action_commit.take_updates();
881
882        let new_schema = match &updates[0] {
883            TableUpdate::AddSchema { schema } => schema,
884            other => panic!("expected AddSchema, got {other:?}"),
885        };
886
887        // "email" should be nested under "person" with ID = last_column_id + 1 = 15.
888        let email = new_schema
889            .field_by_name("person.email")
890            .expect("person.email should exist");
891        assert_eq!(email.id, 15);
892        assert!(!email.required);
893        assert_eq!(*email.field_type, Type::Primitive(PrimitiveType::String));
894
895        // Original nested fields should still be there.
896        assert!(new_schema.field_by_name("person.name").is_some());
897        assert!(new_schema.field_by_name("person.age").is_some());
898    }
899
900    #[tokio::test]
901    async fn test_add_column_to_struct_with_doc() {
902        let table = make_v2_table_with_nested();
903        let tx = Transaction::new(&table);
904
905        let action = tx.update_schema().add_column(
906            AddColumn::builder()
907                .name("phone")
908                .field_type(Type::Primitive(PrimitiveType::String))
909                .parent("person")
910                .doc("Phone number")
911                .build(),
912        );
913
914        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
915        let updates = action_commit.take_updates();
916
917        let new_schema = match &updates[0] {
918            TableUpdate::AddSchema { schema } => schema,
919            other => panic!("expected AddSchema, got {other:?}"),
920        };
921
922        let phone = new_schema
923            .field_by_name("person.phone")
924            .expect("person.phone should exist");
925        assert_eq!(phone.id, 15);
926        assert_eq!(phone.doc.as_deref(), Some("Phone number"));
927    }
928
929    #[tokio::test]
930    async fn test_add_column_to_list_element_struct() {
931        let table = make_v2_table_with_nested();
932        let tx = Transaction::new(&table);
933
934        // "tags" is a list<struct{key, value}>. Adding to the list navigates to its
935        // element struct automatically.
936        let action = tx.update_schema().add_column(
937            AddColumn::builder()
938                .name("score")
939                .field_type(Type::Primitive(PrimitiveType::Double))
940                .parent("tags")
941                .build(),
942        );
943
944        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
945        let updates = action_commit.take_updates();
946
947        let new_schema = match &updates[0] {
948            TableUpdate::AddSchema { schema } => schema,
949            other => panic!("expected AddSchema, got {other:?}"),
950        };
951
952        // The list element struct should now contain "score".
953        let score = new_schema
954            .field_by_name("tags.element.score")
955            .expect("tags.element.score should exist");
956        assert_eq!(score.id, 15);
957        assert!(!score.required);
958
959        // Existing fields preserved.
960        assert!(new_schema.field_by_name("tags.element.key").is_some());
961        assert!(new_schema.field_by_name("tags.element.value").is_some());
962    }
963
964    #[tokio::test]
965    async fn test_add_column_to_map_value_struct() {
966        let table = make_v2_table_with_nested();
967        let tx = Transaction::new(&table);
968
969        // "props" is a map<string, struct{data}>. Adding to the map navigates to its
970        // value struct automatically.
971        let action = tx.update_schema().add_column(
972            AddColumn::builder()
973                .name("version")
974                .field_type(Type::Primitive(PrimitiveType::Int))
975                .parent("props")
976                .build(),
977        );
978
979        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
980        let updates = action_commit.take_updates();
981
982        let new_schema = match &updates[0] {
983            TableUpdate::AddSchema { schema } => schema,
984            other => panic!("expected AddSchema, got {other:?}"),
985        };
986
987        let version = new_schema
988            .field_by_name("props.value.version")
989            .expect("props.value.version should exist");
990        assert_eq!(version.id, 15);
991
992        // Existing map value fields preserved.
993        assert!(new_schema.field_by_name("props.value.data").is_some());
994    }
995
996    #[tokio::test]
997    async fn test_add_column_to_nonexistent_parent_fails() {
998        let table = make_v2_table_with_nested();
999        let tx = Transaction::new(&table);
1000
1001        let action = tx.update_schema().add_column(
1002            AddColumn::builder()
1003                .name("col")
1004                .field_type(Type::Primitive(PrimitiveType::Int))
1005                .parent("nonexistent")
1006                .build(),
1007        );
1008
1009        let err = match Arc::new(action).commit(&table).await {
1010            Err(e) => e,
1011            Ok(_) => panic!("should reject adding to a nonexistent parent"),
1012        };
1013        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
1014        assert!(
1015            err.message().contains("nonexistent"),
1016            "error should mention the missing parent, got: {}",
1017            err.message()
1018        );
1019    }
1020
1021    #[tokio::test]
1022    async fn test_add_column_to_primitive_parent_fails() {
1023        let table = make_v2_table_with_nested();
1024        let tx = Transaction::new(&table);
1025
1026        // "x" is a primitive (long), not a struct.
1027        let action = tx.update_schema().add_column(
1028            AddColumn::builder()
1029                .name("col")
1030                .field_type(Type::Primitive(PrimitiveType::Int))
1031                .parent("x")
1032                .build(),
1033        );
1034
1035        let err = match Arc::new(action).commit(&table).await {
1036            Err(e) => e,
1037            Ok(_) => panic!("should reject adding to a primitive parent"),
1038        };
1039        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
1040        assert!(
1041            err.message().contains("not a struct"),
1042            "error should mention type mismatch, got: {}",
1043            err.message()
1044        );
1045    }
1046
1047    #[tokio::test]
1048    async fn test_add_column_to_nested_name_conflict_fails() {
1049        let table = make_v2_table_with_nested();
1050        let tx = Transaction::new(&table);
1051
1052        // "name" already exists in the "person" struct.
1053        let action = tx.update_schema().add_column(
1054            AddColumn::builder()
1055                .name("name")
1056                .field_type(Type::Primitive(PrimitiveType::String))
1057                .parent("person")
1058                .build(),
1059        );
1060
1061        let err = match Arc::new(action).commit(&table).await {
1062            Err(e) => e,
1063            Ok(_) => panic!("should reject adding a column with conflicting name"),
1064        };
1065        assert_eq!(err.kind(), ErrorKind::PreconditionFailed);
1066        assert!(
1067            err.message().contains("already exists"),
1068            "error should mention name conflict, got: {}",
1069            err.message()
1070        );
1071    }
1072
1073    #[tokio::test]
1074    async fn test_root_and_nested_add_combined() {
1075        let table = make_v2_table_with_nested();
1076        let tx = Transaction::new(&table);
1077
1078        // Add a root column and a nested column in the same action.
1079        let action = tx
1080            .update_schema()
1081            .add_column(AddColumn::optional(
1082                "root_col",
1083                Type::Primitive(PrimitiveType::Boolean),
1084            ))
1085            .add_column(
1086                AddColumn::builder()
1087                    .name("email")
1088                    .field_type(Type::Primitive(PrimitiveType::String))
1089                    .parent("person")
1090                    .build(),
1091            );
1092
1093        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
1094        let updates = action_commit.take_updates();
1095
1096        let new_schema = match &updates[0] {
1097            TableUpdate::AddSchema { schema } => schema,
1098            other => panic!("expected AddSchema, got {other:?}"),
1099        };
1100
1101        // Root column gets the first fresh ID.
1102        let root_col = new_schema
1103            .field_by_name("root_col")
1104            .expect("root_col should exist");
1105        assert_eq!(root_col.id, 15);
1106
1107        // Nested column gets the next ID.
1108        let email = new_schema
1109            .field_by_name("person.email")
1110            .expect("person.email should exist");
1111        assert_eq!(email.id, 16);
1112    }
1113
1114    #[tokio::test]
1115    async fn test_add_nested_struct_type_with_fresh_ids() {
1116        // Adding a new column whose TYPE contains nested fields (e.g. a struct column). All sub-fields must receive
1117        // fresh IDs, not placeholder `DEFAULT_FIELD_ID`.
1118        let table = make_v2_table();
1119        let tx = Transaction::new(&table);
1120
1121        let action = tx.update_schema().add_column(AddColumn::optional(
1122            "address",
1123            Type::Struct(StructType::new(vec![
1124                NestedField::optional(
1125                    DEFAULT_FIELD_ID,
1126                    "street",
1127                    Type::Primitive(PrimitiveType::String),
1128                )
1129                .into(),
1130                NestedField::optional(
1131                    DEFAULT_FIELD_ID,
1132                    "city",
1133                    Type::Primitive(PrimitiveType::String),
1134                )
1135                .into(),
1136            ])),
1137        ));
1138
1139        let mut action_commit = Arc::new(action).commit(&table).await.unwrap();
1140        let updates = action_commit.take_updates();
1141
1142        let new_schema = match &updates[0] {
1143            TableUpdate::AddSchema { schema } => schema,
1144            other => panic!("expected AddSchema, got {other:?}"),
1145        };
1146
1147        // "address" gets ID 4 (last_column_id=3, +1).
1148        let address = new_schema
1149            .field_by_name("address")
1150            .expect("address should exist");
1151        assert_eq!(address.id, 4);
1152
1153        // Sub-fields get IDs 5 and 6.
1154        let street = new_schema
1155            .field_by_name("address.street")
1156            .expect("address.street should exist");
1157        assert_eq!(street.id, 5);
1158
1159        let city = new_schema
1160            .field_by_name("address.city")
1161            .expect("address.city should exist");
1162        assert_eq!(city.id, 6);
1163    }
1164}
iceberg/transaction/update_schema.rs

iceberg/transaction/
update_schema.rs
