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

iceberg/transaction/
update_schema.rs
