Skip to main content

iceberg/spec/
table_metadata_builder.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 uuid::Uuid;
22
23use super::{
24    DEFAULT_PARTITION_SPEC_ID, DEFAULT_SCHEMA_ID, FormatVersion, MAIN_BRANCH, MetadataLog,
25    ONE_MINUTE_MS, PartitionField, PartitionSpec, PartitionSpecBuilder, PartitionStatisticsFile,
26    Schema, SchemaRef, Snapshot, SnapshotLog, SnapshotReference, SnapshotRetention, SortOrder,
27    SortOrderRef, StatisticsFile, StructType, TableMetadata, TableProperties, Transform,
28    UNPARTITIONED_LAST_ASSIGNED_ID, UnboundPartitionSpec,
29};
30use crate::error::{Error, ErrorKind, Result, invalid_data};
31use crate::spec::{EncryptedKey, INITIAL_ROW_ID, MIN_FORMAT_VERSION_ROW_LINEAGE};
32use crate::{TableCreation, TableUpdate};
33
34pub(crate) const FIRST_FIELD_ID: i32 = 1;
35
36/// Manipulating table metadata.
37///
38/// For this builder the order of called functions matters. Functions are applied in-order.
39/// All operations applied to the `TableMetadata` are tracked in `changes` as  a chronologically
40/// ordered vec of `TableUpdate`.
41/// If an operation does not lead to a change of the `TableMetadata`, the corresponding update
42/// is omitted from `changes`.
43///
44/// Unlike a typical builder pattern, the order of function calls matters.
45/// Some basic rules:
46/// - `add_schema` must be called before `set_current_schema`.
47/// - If a new partition spec and schema are added, the schema should be added first.
48#[derive(Debug, Clone)]
49pub struct TableMetadataBuilder {
50    metadata: TableMetadata,
51    changes: Vec<TableUpdate>,
52    last_added_schema_id: Option<i32>,
53    last_added_spec_id: Option<i32>,
54    last_added_order_id: Option<i64>,
55    // None if this is a new table (from_metadata) method not used
56    previous_history_entry: Option<MetadataLog>,
57    last_updated_ms: Option<i64>,
58}
59
60#[derive(Debug, Clone, PartialEq)]
61/// Result of modifying or creating a `TableMetadata`.
62pub struct TableMetadataBuildResult {
63    /// The new `TableMetadata`.
64    pub metadata: TableMetadata,
65    /// The changes that were applied to the metadata.
66    pub changes: Vec<TableUpdate>,
67    /// Expired metadata logs
68    pub expired_metadata_logs: Vec<MetadataLog>,
69}
70
71impl TableMetadataBuilder {
72    /// Proxy id for "last added" items, including schema, partition spec, sort order.
73    pub const LAST_ADDED: i32 = -1;
74
75    /// Create a `TableMetadata` object from scratch.
76    ///
77    /// This method re-assign ids of fields in the schema, schema.id, sort_order.id and
78    /// spec.id. It should only be used to create new table metadata from scratch.
79    pub fn new(
80        schema: Schema,
81        spec: impl Into<UnboundPartitionSpec>,
82        sort_order: SortOrder,
83        location: String,
84        format_version: FormatVersion,
85        properties: HashMap<String, String>,
86    ) -> Result<Self> {
87        // Re-assign field_ids, schema.id, sort_order.id and spec.id for a new table.
88        let (fresh_schema, fresh_spec, fresh_sort_order) =
89            Self::reassign_ids(schema, spec.into(), sort_order)?;
90        let schema_id = fresh_schema.schema_id();
91
92        let builder = Self {
93            metadata: TableMetadata {
94                format_version,
95                table_uuid: Uuid::now_v7(),
96                location: "".to_string(), // Overwritten immediately by set_location
97                last_sequence_number: 0,
98                last_updated_ms: 0,    // Overwritten by build() if not set before
99                last_column_id: -1,    // Overwritten immediately by add_current_schema
100                current_schema_id: -1, // Overwritten immediately by add_current_schema
101                schemas: HashMap::new(),
102                partition_specs: HashMap::new(),
103                default_spec: Arc::new(
104                    // The spec id (-1) is just a proxy value and can be any negative number.
105                    // 0 would lead to wrong changes in the builder if the provided spec by the user is
106                    // also unpartitioned.
107                    // The `default_spec` value is always replaced at the end of this method by he `add_default_partition_spec`
108                    // method.
109                    PartitionSpec::unpartition_spec().with_spec_id(-1),
110                ), // Overwritten immediately by add_default_partition_spec
111                default_partition_type: StructType::new(vec![]),
112                last_partition_id: UNPARTITIONED_LAST_ASSIGNED_ID,
113                properties: HashMap::new(),
114                current_snapshot_id: None,
115                snapshots: HashMap::new(),
116                snapshot_log: vec![],
117                sort_orders: HashMap::new(),
118                metadata_log: vec![],
119                default_sort_order_id: -1, // Overwritten immediately by add_default_sort_order
120                refs: HashMap::default(),
121                statistics: HashMap::new(),
122                partition_statistics: HashMap::new(),
123                encryption_keys: HashMap::new(),
124                next_row_id: INITIAL_ROW_ID,
125            },
126            last_updated_ms: None,
127            changes: vec![],
128            last_added_schema_id: Some(schema_id),
129            last_added_spec_id: None,
130            last_added_order_id: None,
131            previous_history_entry: None,
132        };
133
134        builder
135            .set_location(location)
136            .add_current_schema(fresh_schema)?
137            .add_default_partition_spec(fresh_spec.into_unbound())?
138            .add_default_sort_order(fresh_sort_order)?
139            .set_properties(properties)
140    }
141
142    /// Creates a new table metadata builder from the given metadata to modify it.
143    /// `current_file_location` is the location where the current version
144    /// of the metadata file is stored. This is used to update the metadata log.
145    /// If `current_file_location` is `None`, the metadata log will not be updated.
146    /// This should only be used to stage-create tables.
147    #[must_use]
148    pub fn new_from_metadata(
149        previous: TableMetadata,
150        current_file_location: Option<String>,
151    ) -> Self {
152        Self {
153            previous_history_entry: current_file_location.map(|l| MetadataLog {
154                metadata_file: l,
155                timestamp_ms: previous.last_updated_ms,
156            }),
157            metadata: previous,
158            changes: Vec::default(),
159            last_added_schema_id: None,
160            last_added_spec_id: None,
161            last_added_order_id: None,
162            last_updated_ms: None,
163        }
164    }
165
166    /// Creates a new table metadata builder from the given table creation.
167    pub fn from_table_creation(table_creation: TableCreation) -> Result<Self> {
168        let TableCreation {
169            name: _,
170            location,
171            schema,
172            partition_spec,
173            sort_order,
174            properties,
175            format_version,
176        } = table_creation;
177
178        let location =
179            location.ok_or_else(|| invalid_data!("Can't create table without location"))?;
180        let partition_spec = partition_spec.unwrap_or(UnboundPartitionSpec {
181            spec_id: None,
182            fields: vec![],
183        });
184
185        Self::new(
186            schema,
187            partition_spec,
188            sort_order.unwrap_or(SortOrder::unsorted_order()),
189            location,
190            format_version,
191            properties,
192        )
193    }
194
195    /// Changes uuid of table metadata.
196    pub fn assign_uuid(mut self, uuid: Uuid) -> Self {
197        if self.metadata.table_uuid != uuid {
198            self.metadata.table_uuid = uuid;
199            self.changes.push(TableUpdate::AssignUuid { uuid });
200        }
201
202        self
203    }
204
205    /// Upgrade `FormatVersion`. Downgrades are not allowed.
206    ///
207    /// # Errors
208    /// - Cannot downgrade to older format versions.
209    pub fn upgrade_format_version(mut self, format_version: FormatVersion) -> Result<Self> {
210        if format_version < self.metadata.format_version {
211            return Err(invalid_data!(
212                "Cannot downgrade FormatVersion from {} to {}",
213                self.metadata.format_version,
214                format_version
215            ));
216        }
217
218        if format_version != self.metadata.format_version {
219            match format_version {
220                FormatVersion::V1 => {
221                    // No changes needed for V1
222                }
223                FormatVersion::V2 => {
224                    self.metadata.format_version = format_version;
225                    self.changes
226                        .push(TableUpdate::UpgradeFormatVersion { format_version });
227                }
228                FormatVersion::V3 => {
229                    self.metadata.format_version = format_version;
230                    // Set next-row-id to 0 when upgrading to v3 as per Iceberg spec
231                    self.metadata.next_row_id = INITIAL_ROW_ID;
232                    self.changes
233                        .push(TableUpdate::UpgradeFormatVersion { format_version });
234                }
235            }
236        }
237
238        Ok(self)
239    }
240
241    /// Set properties. If a property already exists, it will be overwritten.
242    ///
243    /// If a reserved property is set, the corresponding action is performed and the property is not persisted.
244    /// Currently the following reserved properties are supported:
245    /// * format-version: Set the format version of the table.
246    ///
247    /// # Errors
248    /// - If properties contains a reserved property
249    pub fn set_properties(mut self, properties: HashMap<String, String>) -> Result<Self> {
250        // List of specified properties that are RESERVED and should not be persisted.
251        let reserved_properties = properties
252            .keys()
253            .filter(|key| TableProperties::RESERVED_PROPERTIES.contains(&key.as_str()))
254            .map(ToString::to_string)
255            .collect::<Vec<_>>();
256
257        if !reserved_properties.is_empty() {
258            return Err(invalid_data!(
259                "Table properties should not contain reserved properties, but got: [{}]",
260                reserved_properties.join(", ")
261            ));
262        }
263
264        if properties.is_empty() {
265            return Ok(self);
266        }
267
268        self.metadata.properties.extend(properties.clone());
269        self.changes.push(TableUpdate::SetProperties {
270            updates: properties,
271        });
272
273        Ok(self)
274    }
275
276    /// Remove properties from the table metadata.
277    /// Does nothing if the key is not present.
278    ///
279    /// # Errors
280    /// - If properties to remove contains a reserved property
281    pub fn remove_properties(mut self, properties: &[String]) -> Result<Self> {
282        // remove duplicates
283        let properties = properties.iter().cloned().collect::<HashSet<_>>();
284
285        // disallow removal of reserved properties
286        let reserved_properties = properties
287            .iter()
288            .filter(|key| TableProperties::RESERVED_PROPERTIES.contains(&key.as_str()))
289            .map(ToString::to_string)
290            .collect::<Vec<_>>();
291
292        if !reserved_properties.is_empty() {
293            return Err(invalid_data!(
294                "Table properties to remove contain reserved properties: [{}]",
295                reserved_properties.join(", ")
296            ));
297        }
298
299        for property in &properties {
300            self.metadata.properties.remove(property);
301        }
302
303        if !properties.is_empty() {
304            self.changes.push(TableUpdate::RemoveProperties {
305                removals: properties.into_iter().collect(),
306            });
307        }
308
309        Ok(self)
310    }
311
312    /// Set the location of the table, stripping any trailing slashes.
313    pub fn set_location(mut self, location: String) -> Self {
314        let location = location.trim_end_matches('/').to_string();
315        if self.metadata.location != location {
316            self.changes.push(TableUpdate::SetLocation {
317                location: location.clone(),
318            });
319            self.metadata.location = location;
320        }
321
322        self
323    }
324
325    /// Add a snapshot to the table metadata.
326    ///
327    /// # Errors
328    /// - Snapshot id already exists.
329    /// - For format version > 1: the sequence number of the snapshot is lower than the highest sequence number specified so far.
330    /// - For format version >= 3: the first-row-id of the snapshot is lower than the next-row-id of the table.
331    /// - For format version >= 3: added-rows is null or first-row-id is null.
332    /// - For format version >= 3: next-row-id would overflow when adding added-rows.
333    pub fn add_snapshot(mut self, snapshot: Snapshot) -> Result<Self> {
334        if self
335            .metadata
336            .snapshots
337            .contains_key(&snapshot.snapshot_id())
338        {
339            return Err(invalid_data!(
340                "Snapshot already exists for: '{}'",
341                snapshot.snapshot_id()
342            ));
343        }
344
345        if self.metadata.format_version != FormatVersion::V1
346            && snapshot.sequence_number() <= self.metadata.last_sequence_number
347            && snapshot.parent_snapshot_id().is_some()
348        {
349            return Err(invalid_data!(
350                "Cannot add snapshot with sequence number {} older than last sequence number {}",
351                snapshot.sequence_number(),
352                self.metadata.last_sequence_number
353            ));
354        }
355
356        if let Some(last) = self.metadata.snapshot_log.last() {
357            // commits can happen concurrently from different machines.
358            // A tolerance helps us avoid failure for small clock skew
359            if snapshot.timestamp_ms() - last.timestamp_ms < -ONE_MINUTE_MS {
360                return Err(invalid_data!(
361                    "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
362                    snapshot.timestamp_ms(),
363                    last.timestamp_ms
364                ));
365            }
366        }
367
368        let max_last_updated = self
369            .last_updated_ms
370            .unwrap_or_default()
371            .max(self.metadata.last_updated_ms);
372        if snapshot.timestamp_ms() - max_last_updated < -ONE_MINUTE_MS {
373            return Err(invalid_data!(
374                "Invalid snapshot timestamp {}: before last updated timestamp {}",
375                snapshot.timestamp_ms(),
376                max_last_updated
377            ));
378        }
379
380        let mut added_rows = None;
381        if self.metadata.format_version >= MIN_FORMAT_VERSION_ROW_LINEAGE {
382            if let Some((first_row_id, added_rows_count)) = snapshot.row_range() {
383                if first_row_id < self.metadata.next_row_id {
384                    return Err(invalid_data!(
385                        "Cannot add a snapshot, first-row-id is behind table next-row-id: {first_row_id} < {}",
386                        self.metadata.next_row_id
387                    ));
388                }
389
390                added_rows = Some(added_rows_count);
391            } else {
392                return Err(invalid_data!(
393                    "Cannot add a snapshot: first-row-id is null. first-row-id must be set for format version >= {MIN_FORMAT_VERSION_ROW_LINEAGE}",
394                ));
395            }
396        }
397
398        if let Some(added_rows) = added_rows {
399            self.metadata.next_row_id = self
400                .metadata
401                .next_row_id
402                .checked_add(added_rows)
403                .ok_or_else(|| {
404                    invalid_data!(
405                        "Cannot add snapshot: next-row-id overflowed when adding added-rows"
406                    )
407                })?;
408        }
409
410        // Mutation happens in next line - must be infallible from here
411        self.changes.push(TableUpdate::AddSnapshot {
412            snapshot: snapshot.clone(),
413        });
414
415        self.last_updated_ms = Some(snapshot.timestamp_ms());
416        self.metadata.last_sequence_number = snapshot.sequence_number();
417        self.metadata
418            .snapshots
419            .insert(snapshot.snapshot_id(), snapshot.into());
420
421        Ok(self)
422    }
423
424    /// Append a snapshot to the specified branch.
425    /// Retention settings from the `branch` are re-used.
426    ///
427    /// # Errors
428    /// - Any of the preconditions of `self.add_snapshot` are not met.
429    pub fn set_branch_snapshot(self, snapshot: Snapshot, branch: &str) -> Result<Self> {
430        let reference = self.metadata.refs.get(branch).cloned();
431
432        let reference = if let Some(mut reference) = reference {
433            if !reference.is_branch() {
434                return Err(invalid_data!(
435                    "Cannot append snapshot to non-branch reference '{branch}'",
436                ));
437            }
438
439            reference.snapshot_id = snapshot.snapshot_id();
440            reference
441        } else {
442            SnapshotReference {
443                snapshot_id: snapshot.snapshot_id(),
444                retention: SnapshotRetention::Branch {
445                    min_snapshots_to_keep: None,
446                    max_snapshot_age_ms: None,
447                    max_ref_age_ms: None,
448                },
449            }
450        };
451
452        self.add_snapshot(snapshot)?.set_ref(branch, reference)
453    }
454
455    /// Remove snapshots by its ids from the table metadata.
456    /// Does nothing if a snapshot id is not present.
457    /// Keeps as changes only the snapshots that were actually removed.
458    pub fn remove_snapshots(mut self, snapshot_ids: &[i64]) -> Self {
459        let mut removed_snapshots = Vec::with_capacity(snapshot_ids.len());
460
461        self.metadata.snapshots.retain(|k, _| {
462            if snapshot_ids.contains(k) {
463                removed_snapshots.push(*k);
464                false
465            } else {
466                true
467            }
468        });
469
470        if !removed_snapshots.is_empty() {
471            self.changes.push(TableUpdate::RemoveSnapshots {
472                snapshot_ids: removed_snapshots,
473            });
474        }
475
476        // Remove refs that are no longer valid
477        self.metadata
478            .refs
479            .retain(|_, v| self.metadata.snapshots.contains_key(&v.snapshot_id));
480
481        self
482    }
483
484    /// Set a reference to a snapshot.
485    ///
486    /// # Errors
487    /// - The snapshot id is unknown.
488    pub fn set_ref(mut self, ref_name: &str, reference: SnapshotReference) -> Result<Self> {
489        if self
490            .metadata
491            .refs
492            .get(ref_name)
493            .is_some_and(|snap_ref| snap_ref.eq(&reference))
494        {
495            return Ok(self);
496        }
497
498        let Some(snapshot) = self.metadata.snapshots.get(&reference.snapshot_id) else {
499            return Err(invalid_data!(
500                "Cannot set '{ref_name}' to unknown snapshot: '{}'",
501                reference.snapshot_id
502            ));
503        };
504
505        // Update last_updated_ms to the exact timestamp of the snapshot if it was added in this commit
506        let is_added_snapshot = self.changes.iter().any(|update| {
507            matches!(update, TableUpdate::AddSnapshot { snapshot: snap } if snap.snapshot_id() == snapshot.snapshot_id())
508        });
509        if is_added_snapshot {
510            self.last_updated_ms = Some(snapshot.timestamp_ms());
511        }
512
513        // Current snapshot id is set only for the main branch
514        if ref_name == MAIN_BRANCH {
515            self.metadata.current_snapshot_id = Some(snapshot.snapshot_id());
516            let timestamp_ms = if let Some(last_updated_ms) = self.last_updated_ms {
517                last_updated_ms
518            } else {
519                let last_updated_ms = chrono::Utc::now().timestamp_millis();
520                self.last_updated_ms = Some(last_updated_ms);
521                last_updated_ms
522            };
523
524            self.metadata.snapshot_log.push(SnapshotLog {
525                snapshot_id: snapshot.snapshot_id(),
526                timestamp_ms,
527            });
528        }
529
530        self.changes.push(TableUpdate::SetSnapshotRef {
531            ref_name: ref_name.to_string(),
532            reference: reference.clone(),
533        });
534        self.metadata.refs.insert(ref_name.to_string(), reference);
535
536        Ok(self)
537    }
538
539    /// Remove a reference
540    ///
541    /// If `ref_name='main'` the current snapshot id is set to `None`.
542    pub fn remove_ref(mut self, ref_name: &str) -> Self {
543        if ref_name == MAIN_BRANCH {
544            self.metadata.current_snapshot_id = None;
545        }
546
547        if self.metadata.refs.remove(ref_name).is_some() || ref_name == MAIN_BRANCH {
548            self.changes.push(TableUpdate::RemoveSnapshotRef {
549                ref_name: ref_name.to_string(),
550            });
551        }
552
553        self
554    }
555
556    /// Set statistics for a snapshot
557    pub fn set_statistics(mut self, statistics: StatisticsFile) -> Self {
558        self.metadata
559            .statistics
560            .insert(statistics.snapshot_id, statistics.clone());
561        self.changes.push(TableUpdate::SetStatistics {
562            statistics: statistics.clone(),
563        });
564        self
565    }
566
567    /// Remove statistics for a snapshot
568    pub fn remove_statistics(mut self, snapshot_id: i64) -> Self {
569        let previous = self.metadata.statistics.remove(&snapshot_id);
570        if previous.is_some() {
571            self.changes
572                .push(TableUpdate::RemoveStatistics { snapshot_id });
573        }
574        self
575    }
576
577    /// Set partition statistics
578    pub fn set_partition_statistics(
579        mut self,
580        partition_statistics_file: PartitionStatisticsFile,
581    ) -> Self {
582        self.metadata.partition_statistics.insert(
583            partition_statistics_file.snapshot_id,
584            partition_statistics_file.clone(),
585        );
586        self.changes.push(TableUpdate::SetPartitionStatistics {
587            partition_statistics: partition_statistics_file,
588        });
589        self
590    }
591
592    /// Remove partition statistics
593    pub fn remove_partition_statistics(mut self, snapshot_id: i64) -> Self {
594        let previous = self.metadata.partition_statistics.remove(&snapshot_id);
595        if previous.is_some() {
596            self.changes
597                .push(TableUpdate::RemovePartitionStatistics { snapshot_id });
598        }
599        self
600    }
601
602    /// Add a schema to the table metadata.
603    ///
604    /// The provided `schema.schema_id` may not be used.
605    ///
606    /// Important: Use this method with caution. The builder does not check
607    /// if the added schema is compatible with the current schema.
608    pub fn add_schema(mut self, schema: Schema) -> Result<Self> {
609        let new_schema_id = self.reuse_or_create_new_schema_id(&schema);
610        let schema_found = self.metadata.schemas.contains_key(&new_schema_id);
611
612        if schema_found {
613            if self.last_added_schema_id != Some(new_schema_id) {
614                self.changes.push(TableUpdate::AddSchema {
615                    schema: schema.clone(),
616                });
617                self.last_added_schema_id = Some(new_schema_id);
618            }
619
620            return Ok(self);
621        }
622
623        // New schemas might contain only old columns. In this case last_column_id should not be
624        // reduced.
625        self.metadata.last_column_id =
626            std::cmp::max(self.metadata.last_column_id, schema.highest_field_id());
627
628        // Set schema-id
629        let schema = match new_schema_id == schema.schema_id() {
630            true => schema,
631            false => schema.with_schema_id(new_schema_id),
632        };
633
634        self.metadata
635            .schemas
636            .insert(new_schema_id, schema.clone().into());
637
638        self.changes.push(TableUpdate::AddSchema { schema });
639
640        self.last_added_schema_id = Some(new_schema_id);
641
642        Ok(self)
643    }
644
645    /// Set the current schema id.
646    ///
647    /// If `schema_id` is -1, the last added schema is set as the current schema.
648    ///
649    /// Errors:
650    /// - provided `schema_id` is -1 but no schema has been added via `add_schema`.
651    /// - No schema with the provided `schema_id` exists.
652    /// - A field of the schema takes the name of a partition field it is not the source of.
653    pub fn set_current_schema(mut self, mut schema_id: i32) -> Result<Self> {
654        if schema_id == Self::LAST_ADDED {
655            schema_id = self.last_added_schema_id.ok_or_else(|| {
656                invalid_data!(
657                    "Cannot set current schema to last added schema: no schema has been added."
658                )
659            })?;
660        };
661        let schema_id = schema_id; // Make immutable
662
663        if schema_id == self.metadata.current_schema_id {
664            return Ok(self);
665        }
666
667        let schema = self.metadata.schemas.get(&schema_id).ok_or_else(|| {
668            invalid_data!("Cannot set current schema to unknown schema with id: '{schema_id}'")
669        })?;
670        self.validate_schema_field_names(schema)?;
671
672        // Old partition specs and sort-orders should be preserved even if they are not compatible with the new schema,
673        // so that older metadata can still be interpreted.
674        // Default partition spec and sort order are checked in the build() method
675        // which allows other default partition specs and sort orders to be set before the build.
676
677        self.metadata.current_schema_id = schema_id;
678
679        if self.last_added_schema_id == Some(schema_id) {
680            self.changes.push(TableUpdate::SetCurrentSchema {
681                schema_id: Self::LAST_ADDED,
682            });
683        } else {
684            self.changes
685                .push(TableUpdate::SetCurrentSchema { schema_id });
686        }
687
688        Ok(self)
689    }
690
691    /// Add a schema and set it as the current schema.
692    pub fn add_current_schema(self, schema: Schema) -> Result<Self> {
693        self.add_schema(schema)?
694            .set_current_schema(Self::LAST_ADDED)
695    }
696
697    /// Validate schema field names against the partition field names of every spec.
698    ///
699    /// A partition field whose source column is absent from `schema` is skipped: its spec is
700    /// historical, kept only so old metadata can be interpreted, so the name is free to reuse.
701    /// Otherwise the name may only be taken by the source column itself, which is the identity
702    /// partition case that [`Self::validate_partition_field_names`] allows in the other direction.
703    ///
704    /// # Errors
705    /// - A schema field takes the name of a partition field it is not the source of.
706    fn validate_schema_field_names(&self, schema: &Schema) -> Result<()> {
707        for (spec_id, spec) in &self.metadata.partition_specs {
708            for partition_field in spec.fields() {
709                if schema.field_by_id(partition_field.source_id).is_none() {
710                    continue;
711                }
712
713                let Some(schema_field) = schema.field_by_name(&partition_field.name) else {
714                    continue;
715                };
716
717                if schema_field.id != partition_field.source_id {
718                    return Err(invalid_data!(
719                        "Cannot set current schema {}: field '{}' with id {} conflicts with a \
720                         partition field of that name in spec {} sourced from field id {}",
721                        schema.schema_id(),
722                        partition_field.name,
723                        schema_field.id,
724                        spec_id,
725                        partition_field.source_id
726                    ));
727                }
728            }
729        }
730
731        Ok(())
732    }
733
734    /// Validate partition field names against schema field names across all historical schemas.
735    ///
736    /// Due to Iceberg's multi-version property, partition fields can share names with schema fields
737    /// if they meet specific requirements (identity or void transform + matching source field ID).
738    /// This validation enforces those rules across all historical schema versions.
739    ///
740    /// # Errors
741    /// - Partition field name conflicts with schema field name but doesn't use identity or void
742    ///   transform.
743    /// - Partition field uses identity or void transform but references wrong source field ID.
744    fn validate_partition_field_names(&self, unbound_spec: &UnboundPartitionSpec) -> Result<()> {
745        if self.metadata.schemas.is_empty() {
746            return Ok(());
747        }
748
749        let current_schema = self.get_current_schema()?;
750        for partition_field in unbound_spec.fields() {
751            let exists_in_any_schema = self
752                .metadata
753                .name_exists_in_any_schema(partition_field.name());
754
755            // Skip if partition field name doesn't conflict with any schema field
756            if !exists_in_any_schema {
757                continue;
758            }
759
760            // If name exists in schemas, validate against current schema rules
761            if let Some(schema_field) = current_schema.field_by_name(partition_field.name()) {
762                let has_matching_source_id = schema_field.id == partition_field.source_id()?;
763
764                if !partition_field.transform().may_share_name_with_source() {
765                    return Err(invalid_data!(
766                        "Cannot create partition with name '{}' that conflicts with schema field and is not an identity or void transform.",
767                        partition_field.name()
768                    ));
769                }
770
771                if !has_matching_source_id {
772                    return Err(invalid_data!(
773                        "Cannot create identity or void partition sourced from different field in schema. \
774                             Field name '{}' has id `{}` in schema but partition source id is `{}`",
775                        partition_field.name(),
776                        schema_field.id,
777                        partition_field.source_id()?
778                    ));
779                }
780            }
781        }
782
783        Ok(())
784    }
785
786    /// Add a partition spec to the table metadata.
787    ///
788    /// The spec is bound eagerly to the current schema.
789    /// If a schema is added in the same set of changes, the schema should be added first.
790    ///
791    /// Even if `unbound_spec.spec_id` is provided as `Some`, it may not be used.
792    ///
793    /// # Errors
794    /// - The partition spec cannot be bound to the current schema.
795    /// - The partition spec has non-sequential field ids and the table format version is 1.
796    pub fn add_partition_spec(mut self, unbound_spec: UnboundPartitionSpec) -> Result<Self> {
797        let schema = self.get_current_schema()?.clone();
798
799        // Check if partition field names conflict with schema field names across all schemas
800        self.validate_partition_field_names(&unbound_spec)?;
801
802        // Reuse field IDs for equivalent fields from existing partition specs
803        let unbound_spec = self.reuse_partition_field_ids(unbound_spec)?;
804
805        let spec = PartitionSpecBuilder::new_from_unbound(unbound_spec.clone(), schema)?
806            .with_last_assigned_field_id(self.metadata.last_partition_id)
807            .build()?;
808
809        let new_spec_id = self.reuse_or_create_new_spec_id(&spec);
810        let spec_found = self.metadata.partition_specs.contains_key(&new_spec_id);
811        let spec = spec.with_spec_id(new_spec_id);
812        let unbound_spec = unbound_spec.with_spec_id(new_spec_id);
813
814        if spec_found {
815            if self.last_added_spec_id != Some(new_spec_id) {
816                self.changes
817                    .push(TableUpdate::AddSpec { spec: unbound_spec });
818                self.last_added_spec_id = Some(new_spec_id);
819            }
820
821            return Ok(self);
822        }
823
824        if self.metadata.format_version <= FormatVersion::V1 && !spec.has_sequential_ids() {
825            return Err(invalid_data!(
826                "Cannot add partition spec with non-sequential field ids to format version 1 table"
827            ));
828        }
829
830        let highest_field_id = spec
831            .highest_field_id()
832            .unwrap_or(UNPARTITIONED_LAST_ASSIGNED_ID);
833        self.metadata
834            .partition_specs
835            .insert(new_spec_id, Arc::new(spec));
836        self.changes
837            .push(TableUpdate::AddSpec { spec: unbound_spec });
838
839        self.last_added_spec_id = Some(new_spec_id);
840        self.metadata.last_partition_id =
841            std::cmp::max(self.metadata.last_partition_id, highest_field_id);
842
843        Ok(self)
844    }
845
846    /// Reuse partition field IDs for equivalent fields from existing partition specs.
847    ///
848    /// According to the Iceberg spec, partition field IDs must be reused if an existing
849    /// partition spec contains an equivalent field (same source_id and transform).
850    fn reuse_partition_field_ids(
851        &self,
852        unbound_spec: UnboundPartitionSpec,
853    ) -> Result<UnboundPartitionSpec> {
854        let equivalent_field_ids = Self::highest_equivalent_field_ids(
855            self.metadata
856                .partition_specs
857                .values()
858                .flat_map(|spec| spec.fields()),
859        );
860
861        // Create new fields with reused field IDs where possible
862        let fields = unbound_spec
863            .fields
864            .into_iter()
865            .map(|field| {
866                if field.field_id().is_none()
867                    && let Some(&existing_field_id) =
868                        equivalent_field_ids.get(&(field.source_ids().to_vec(), field.transform()))
869                {
870                    field.with_field_id(existing_field_id)
871                } else {
872                    field
873                }
874            })
875            .collect();
876
877        Ok(UnboundPartitionSpec {
878            spec_id: unbound_spec.spec_id,
879            fields,
880        })
881    }
882
883    /// Map each (source_ids, transform) pair to the highest field id assigned to it.
884    ///
885    /// Taking the highest keeps the result independent of the order `fields` arrive in, which for
886    /// the specs of a `TableMetadata` is an arbitrary `HashMap` iteration order.
887    fn highest_equivalent_field_ids<'a>(
888        fields: impl Iterator<Item = &'a PartitionField>,
889    ) -> HashMap<(Vec<i32>, Transform), i32> {
890        let mut equivalent_field_ids = HashMap::new();
891        for field in fields {
892            equivalent_field_ids
893                .entry((vec![field.source_id], field.transform))
894                .and_modify(|id: &mut i32| *id = (*id).max(field.field_id))
895                .or_insert(field.field_id);
896        }
897        equivalent_field_ids
898    }
899
900    /// Set the default partition spec.
901    ///
902    /// # Errors
903    /// - spec_id is -1 but no spec has been added via `add_partition_spec`.
904    /// - No partition spec with the provided `spec_id` exists.
905    pub fn set_default_partition_spec(mut self, mut spec_id: i32) -> Result<Self> {
906        if spec_id == Self::LAST_ADDED {
907            spec_id = self.last_added_spec_id.ok_or_else(|| {
908                invalid_data!(
909                    "Cannot set default partition spec to last added spec: no spec has been added."
910                )
911            })?;
912        }
913
914        if self.metadata.default_spec.spec_id() == spec_id {
915            return Ok(self);
916        }
917
918        if !self.metadata.partition_specs.contains_key(&spec_id) {
919            return Err(invalid_data!(
920                "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
921            ));
922        }
923
924        let schemaless_spec = self
925            .metadata
926            .partition_specs
927            .get(&spec_id)
928            .ok_or_else(|| {
929                invalid_data!(
930                    "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
931                )
932            })?
933            .clone();
934        let spec = Arc::unwrap_or_clone(schemaless_spec);
935        let spec_type = spec.partition_type(self.get_current_schema()?)?;
936        self.metadata.default_spec = Arc::new(spec);
937        self.metadata.default_partition_type = spec_type;
938
939        if self.last_added_spec_id == Some(spec_id) {
940            self.changes.push(TableUpdate::SetDefaultSpec {
941                spec_id: Self::LAST_ADDED,
942            });
943        } else {
944            self.changes.push(TableUpdate::SetDefaultSpec { spec_id });
945        }
946
947        Ok(self)
948    }
949
950    /// Add a partition spec and set it as the default
951    pub fn add_default_partition_spec(self, unbound_spec: UnboundPartitionSpec) -> Result<Self> {
952        self.add_partition_spec(unbound_spec)?
953            .set_default_partition_spec(Self::LAST_ADDED)
954    }
955
956    /// Remove partition specs by their ids from the table metadata.
957    /// Does nothing if a spec id is not present. Active partition specs
958    /// should not be removed.
959    ///
960    /// # Errors
961    /// - Cannot remove the default partition spec.
962    pub fn remove_partition_specs(mut self, spec_ids: &[i32]) -> Result<Self> {
963        if spec_ids.contains(&self.metadata.default_spec.spec_id()) {
964            return Err(invalid_data!("Cannot remove default partition spec"));
965        }
966
967        let mut removed_specs = Vec::with_capacity(spec_ids.len());
968        spec_ids.iter().for_each(|id| {
969            if self.metadata.partition_specs.remove(id).is_some() {
970                removed_specs.push(*id);
971            }
972        });
973
974        if !removed_specs.is_empty() {
975            self.changes.push(TableUpdate::RemovePartitionSpecs {
976                spec_ids: removed_specs,
977            });
978        }
979
980        Ok(self)
981    }
982
983    /// Add a sort order to the table metadata.
984    ///
985    /// The spec is bound eagerly to the current schema and must be valid for it.
986    /// If a schema is added in the same set of changes, the schema should be added first.
987    ///
988    /// Even if `sort_order.order_id` is provided, it may not be used.
989    ///
990    /// # Errors
991    /// - Sort order id to add already exists.
992    /// - Sort order is incompatible with the current schema.
993    pub fn add_sort_order(mut self, sort_order: SortOrder) -> Result<Self> {
994        let new_order_id = self.reuse_or_create_new_sort_id(&sort_order);
995        let sort_order_found = self.metadata.sort_orders.contains_key(&new_order_id);
996
997        if sort_order_found {
998            if self.last_added_order_id != Some(new_order_id) {
999                self.changes.push(TableUpdate::AddSortOrder {
1000                    sort_order: sort_order.clone().with_order_id(new_order_id),
1001                });
1002                self.last_added_order_id = Some(new_order_id);
1003            }
1004
1005            return Ok(self);
1006        }
1007
1008        let schema = self.get_current_schema()?.clone().as_ref().clone();
1009        let sort_order = SortOrder::builder()
1010            .with_order_id(new_order_id)
1011            .with_fields(sort_order.fields)
1012            .build(&schema)
1013            .map_err(|e| {
1014                invalid_data!("Sort order to add is incompatible with current schema: {e}")
1015                    .with_source(e)
1016            })?;
1017
1018        self.last_added_order_id = Some(new_order_id);
1019        self.metadata
1020            .sort_orders
1021            .insert(new_order_id, sort_order.clone().into());
1022        self.changes.push(TableUpdate::AddSortOrder { sort_order });
1023
1024        Ok(self)
1025    }
1026
1027    /// Set the default sort order. If `sort_order_id` is -1, the last added sort order is set as default.
1028    ///
1029    /// # Errors
1030    /// - sort_order_id is -1 but no sort order has been added via `add_sort_order`.
1031    /// - No sort order with the provided `sort_order_id` exists.
1032    pub fn set_default_sort_order(mut self, mut sort_order_id: i64) -> Result<Self> {
1033        if sort_order_id == Self::LAST_ADDED as i64 {
1034            sort_order_id = self.last_added_order_id.ok_or_else(|| {
1035                invalid_data!(
1036                    "Cannot set default sort order to last added order: no order has been added."
1037                )
1038            })?;
1039        }
1040
1041        if self.metadata.default_sort_order_id == sort_order_id {
1042            return Ok(self);
1043        }
1044
1045        if !self.metadata.sort_orders.contains_key(&sort_order_id) {
1046            return Err(invalid_data!(
1047                "Cannot set default sort order to unknown order with id: '{sort_order_id}'"
1048            ));
1049        }
1050
1051        self.metadata.default_sort_order_id = sort_order_id;
1052
1053        if self.last_added_order_id == Some(sort_order_id) {
1054            self.changes.push(TableUpdate::SetDefaultSortOrder {
1055                sort_order_id: Self::LAST_ADDED as i64,
1056            });
1057        } else {
1058            self.changes
1059                .push(TableUpdate::SetDefaultSortOrder { sort_order_id });
1060        }
1061
1062        Ok(self)
1063    }
1064
1065    /// Add a sort order and set it as the default
1066    fn add_default_sort_order(self, sort_order: SortOrder) -> Result<Self> {
1067        self.add_sort_order(sort_order)?
1068            .set_default_sort_order(Self::LAST_ADDED as i64)
1069    }
1070
1071    /// Add an encryption key to the table metadata.
1072    pub fn add_encryption_key(mut self, key: EncryptedKey) -> Self {
1073        let key_id = key.key_id().to_string();
1074        if self.metadata.encryption_keys.contains_key(&key_id) {
1075            // already exists
1076            return self;
1077        }
1078
1079        self.metadata.encryption_keys.insert(key_id, key.clone());
1080        self.changes.push(TableUpdate::AddEncryptionKey {
1081            encryption_key: key,
1082        });
1083        self
1084    }
1085
1086    /// Remove an encryption key from the table metadata.
1087    pub fn remove_encryption_key(mut self, key_id: &str) -> Self {
1088        if self.metadata.encryption_keys.remove(key_id).is_some() {
1089            self.changes.push(TableUpdate::RemoveEncryptionKey {
1090                key_id: key_id.to_string(),
1091            });
1092        }
1093        self
1094    }
1095
1096    /// Build the table metadata.
1097    pub fn build(mut self) -> Result<TableMetadataBuildResult> {
1098        self.metadata.last_updated_ms = self
1099            .last_updated_ms
1100            .unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
1101
1102        // Check compatibility of the current schema to the default partition spec and sort order.
1103        // We use the `get_xxx` methods from the builder to avoid using the panicking
1104        // `TableMetadata.default_partition_spec` etc. methods.
1105        let schema = self.get_current_schema()?.clone();
1106        let sort_order = Arc::unwrap_or_clone(self.get_default_sort_order()?);
1107
1108        self.metadata.default_spec = Arc::new(
1109            Arc::unwrap_or_clone(self.metadata.default_spec)
1110                .into_unbound()
1111                .bind(schema.clone())?,
1112        );
1113        self.metadata.default_partition_type =
1114            self.metadata.default_spec.partition_type(&schema)?;
1115        SortOrder::builder()
1116            .with_fields(sort_order.fields)
1117            .build(&schema)?;
1118
1119        self.update_snapshot_log()?;
1120        self.metadata.try_normalize()?;
1121
1122        if let Some(hist_entry) = self.previous_history_entry.take() {
1123            self.metadata.metadata_log.push(hist_entry);
1124        }
1125        let expired_metadata_logs = self.expire_metadata_log();
1126
1127        Ok(TableMetadataBuildResult {
1128            metadata: self.metadata,
1129            changes: self.changes,
1130            expired_metadata_logs,
1131        })
1132    }
1133
1134    fn expire_metadata_log(&mut self) -> Vec<MetadataLog> {
1135        let max_size = self
1136            .metadata
1137            .properties
1138            .get(TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX)
1139            .and_then(|v| v.parse::<usize>().ok())
1140            .unwrap_or(TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT)
1141            .max(1);
1142
1143        if self.metadata.metadata_log.len() > max_size {
1144            self.metadata
1145                .metadata_log
1146                .drain(0..self.metadata.metadata_log.len() - max_size)
1147                .collect()
1148        } else {
1149            Vec::new()
1150        }
1151    }
1152
1153    fn update_snapshot_log(&mut self) -> Result<()> {
1154        let intermediate_snapshots = self.get_intermediate_snapshots();
1155        let has_removed_snapshots = self
1156            .changes
1157            .iter()
1158            .any(|update| matches!(update, TableUpdate::RemoveSnapshots { .. }));
1159
1160        if intermediate_snapshots.is_empty() && !has_removed_snapshots {
1161            return Ok(());
1162        }
1163
1164        let mut new_snapshot_log = Vec::new();
1165        for log_entry in &self.metadata.snapshot_log {
1166            let snapshot_id = log_entry.snapshot_id;
1167            if self.metadata.snapshots.contains_key(&snapshot_id) {
1168                if !intermediate_snapshots.contains(&snapshot_id) {
1169                    new_snapshot_log.push(log_entry.clone());
1170                }
1171            } else if has_removed_snapshots {
1172                // any invalid entry causes the history before it to be removed. otherwise, there could be
1173                // history gaps that cause time-travel queries to produce incorrect results. for example,
1174                // if history is [(t1, s1), (t2, s2), (t3, s3)] and s2 is removed, the history cannot be
1175                // [(t1, s1), (t3, s3)] because it appears that s3 was current during the time between t2
1176                // and t3 when in fact s2 was the current snapshot.
1177                new_snapshot_log.clear();
1178            }
1179        }
1180
1181        if let Some(current_snapshot_id) = self.metadata.current_snapshot_id {
1182            let last_id = new_snapshot_log.last().map(|entry| entry.snapshot_id);
1183            if last_id != Some(current_snapshot_id) {
1184                return Err(invalid_data!(
1185                    "Cannot set invalid snapshot log: latest entry is not the current snapshot"
1186                ));
1187            }
1188        };
1189
1190        self.metadata.snapshot_log = new_snapshot_log;
1191        Ok(())
1192    }
1193
1194    /// Finds intermediate snapshots that have not been committed as the current snapshot.
1195    ///
1196    /// Transactions can create snapshots that are never the current snapshot because several
1197    /// changes are combined by the transaction into one table metadata update. when each
1198    /// intermediate snapshot is added to table metadata, it is added to the snapshot log, assuming
1199    /// that it will be the current snapshot. when there are multiple snapshot updates, the log must
1200    /// be corrected by suppressing the intermediate snapshot entries.
1201    ///     
1202    /// A snapshot is an intermediate snapshot if it was added but is not the current snapshot.
1203    fn get_intermediate_snapshots(&self) -> HashSet<i64> {
1204        let added_snapshot_ids = self
1205            .changes
1206            .iter()
1207            .filter_map(|update| match update {
1208                TableUpdate::AddSnapshot { snapshot } => Some(snapshot.snapshot_id()),
1209                _ => None,
1210            })
1211            .collect::<HashSet<_>>();
1212
1213        self.changes
1214            .iter()
1215            .filter_map(|update| match update {
1216                TableUpdate::SetSnapshotRef {
1217                    ref_name,
1218                    reference,
1219                } => {
1220                    if added_snapshot_ids.contains(&reference.snapshot_id)
1221                        && ref_name == MAIN_BRANCH
1222                        && reference.snapshot_id
1223                            != self
1224                                .metadata
1225                                .current_snapshot_id
1226                                .unwrap_or(i64::from(Self::LAST_ADDED))
1227                    {
1228                        Some(reference.snapshot_id)
1229                    } else {
1230                        None
1231                    }
1232                }
1233                _ => None,
1234            })
1235            .collect()
1236    }
1237
1238    fn reassign_ids(
1239        schema: Schema,
1240        spec: UnboundPartitionSpec,
1241        sort_order: SortOrder,
1242    ) -> Result<(Schema, PartitionSpec, SortOrder)> {
1243        // Re-assign field ids and schema ids for a new table.
1244        let previous_id_to_name = schema.field_id_to_name_map().clone();
1245        let fresh_schema = schema
1246            .into_builder()
1247            .with_schema_id(DEFAULT_SCHEMA_ID)
1248            .with_reassigned_field_ids(FIRST_FIELD_ID)
1249            .build()?;
1250
1251        // Re-build partition spec with new ids
1252        let mut fresh_spec = PartitionSpecBuilder::new(fresh_schema.clone());
1253        for field in spec.fields() {
1254            let source_id = field.source_id()?;
1255            let source_field_name = previous_id_to_name.get(&source_id).ok_or_else(|| {
1256                invalid_data!(
1257                    "Cannot find source column with id {} for partition column {} in schema.",
1258                    source_id,
1259                    field.name()
1260                )
1261            })?;
1262            fresh_spec = fresh_spec.add_partition_field(
1263                source_field_name,
1264                field.name(),
1265                field.transform(),
1266            )?;
1267        }
1268        let fresh_spec = fresh_spec.build()?;
1269
1270        // Re-build sort order with new ids
1271        let mut fresh_order = SortOrder::builder();
1272        for mut field in sort_order.fields {
1273            let source_field_name = previous_id_to_name.get(&field.source_id).ok_or_else(|| {
1274                invalid_data!(
1275                    "Cannot find source column with id {} for sort column in schema.",
1276                    field.source_id
1277                )
1278            })?;
1279            let new_field_id = fresh_schema
1280                       .field_by_name(source_field_name)
1281                       .ok_or_else(|| {
1282                           Error::new(
1283                               ErrorKind::Unexpected,
1284                               format!(
1285                                   "Cannot find source column with name {source_field_name} for sort column in re-assigned schema."
1286                               ),
1287                           )
1288                       })?.id;
1289            field.source_id = new_field_id;
1290            fresh_order.with_sort_field(field);
1291        }
1292        let fresh_sort_order = fresh_order.build(&fresh_schema)?;
1293
1294        Ok((fresh_schema, fresh_spec, fresh_sort_order))
1295    }
1296
1297    fn reuse_or_create_new_schema_id(&self, new_schema: &Schema) -> i32 {
1298        self.metadata
1299            .schemas
1300            .iter()
1301            .find_map(|(id, schema)| new_schema.is_same_schema(schema).then_some(*id))
1302            .unwrap_or_else(|| self.get_highest_schema_id() + 1)
1303    }
1304
1305    fn get_highest_schema_id(&self) -> i32 {
1306        *self
1307            .metadata
1308            .schemas
1309            .keys()
1310            .max()
1311            .unwrap_or(&self.metadata.current_schema_id)
1312    }
1313
1314    fn get_current_schema(&self) -> Result<&SchemaRef> {
1315        self.metadata
1316            .schemas
1317            .get(&self.metadata.current_schema_id)
1318            .ok_or_else(|| {
1319                invalid_data!(
1320                    "Current schema with id '{}' not found in table metadata.",
1321                    self.metadata.current_schema_id
1322                )
1323            })
1324    }
1325
1326    fn get_default_sort_order(&self) -> Result<SortOrderRef> {
1327        self.metadata
1328            .sort_orders
1329            .get(&self.metadata.default_sort_order_id)
1330            .cloned()
1331            .ok_or_else(|| {
1332                invalid_data!(
1333                    "Default sort order with id '{}' not found in table metadata.",
1334                    self.metadata.default_sort_order_id
1335                )
1336            })
1337    }
1338
1339    /// If a compatible spec already exists, use the same ID. Otherwise, use 1 more than the highest ID.
1340    fn reuse_or_create_new_spec_id(&self, new_spec: &PartitionSpec) -> i32 {
1341        self.metadata
1342            .partition_specs
1343            .iter()
1344            .find_map(|(id, old_spec)| new_spec.is_compatible_with(old_spec).then_some(*id))
1345            .unwrap_or_else(|| {
1346                self.get_highest_spec_id()
1347                    .map(|id| id + 1)
1348                    .unwrap_or(DEFAULT_PARTITION_SPEC_ID)
1349            })
1350    }
1351
1352    fn get_highest_spec_id(&self) -> Option<i32> {
1353        self.metadata.partition_specs.keys().max().copied()
1354    }
1355
1356    /// If a compatible sort-order already exists, use the same ID. Otherwise, use 1 more than the highest ID.
1357    fn reuse_or_create_new_sort_id(&self, new_sort_order: &SortOrder) -> i64 {
1358        if new_sort_order.is_unsorted() {
1359            return SortOrder::unsorted_order().order_id;
1360        }
1361
1362        self.metadata
1363            .sort_orders
1364            .iter()
1365            .find_map(|(id, sort_order)| {
1366                sort_order.fields.eq(&new_sort_order.fields).then_some(*id)
1367            })
1368            .unwrap_or_else(|| {
1369                self.highest_sort_order_id()
1370                    .unwrap_or(SortOrder::unsorted_order().order_id)
1371                    + 1
1372            })
1373    }
1374
1375    fn highest_sort_order_id(&self) -> Option<i64> {
1376        self.metadata.sort_orders.keys().max().copied()
1377    }
1378
1379    /// Remove schemas by their ids from the table metadata.
1380    /// Does nothing if a schema id is not present. Active schemas should not be removed.
1381    pub fn remove_schemas(mut self, schema_id_to_remove: &[i32]) -> Result<Self> {
1382        if schema_id_to_remove.contains(&self.metadata.current_schema_id) {
1383            return Err(invalid_data!("Cannot remove current schema"));
1384        }
1385
1386        if schema_id_to_remove.is_empty() {
1387            return Ok(self);
1388        }
1389
1390        let mut removed_schemas = Vec::with_capacity(schema_id_to_remove.len());
1391        self.metadata.schemas.retain(|id, _schema| {
1392            if schema_id_to_remove.contains(id) {
1393                removed_schemas.push(*id);
1394                false
1395            } else {
1396                true
1397            }
1398        });
1399
1400        self.changes.push(TableUpdate::RemoveSchemas {
1401            schema_ids: removed_schemas,
1402        });
1403
1404        Ok(self)
1405    }
1406}
1407
1408impl From<TableMetadataBuildResult> for TableMetadata {
1409    fn from(result: TableMetadataBuildResult) -> Self {
1410        result.metadata
1411    }
1412}
1413
1414#[cfg(test)]
1415mod tests {
1416    use std::fs::File;
1417    use std::io::BufReader;
1418    use std::thread::sleep;
1419
1420    use super::*;
1421    use crate::TableIdent;
1422    use crate::io::FileIO;
1423    use crate::spec::{
1424        BlobMetadata, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema,
1425        SnapshotRetention, SortDirection, SortField, StructType, Summary, TableProperties,
1426        Transform, Type, UnboundPartitionField,
1427    };
1428    use crate::table::Table;
1429    use crate::test_utils::test_runtime;
1430
1431    const TEST_LOCATION: &str = "s3://bucket/test/location";
1432    const LAST_ASSIGNED_COLUMN_ID: i32 = 3;
1433
1434    fn schema() -> Schema {
1435        Schema::builder()
1436            .with_fields(vec![
1437                NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
1438                NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
1439                NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
1440            ])
1441            .build()
1442            .unwrap()
1443    }
1444
1445    fn sort_order() -> SortOrder {
1446        let schema = schema();
1447        SortOrder::builder()
1448            .with_order_id(1)
1449            .with_sort_field(SortField {
1450                source_id: 3,
1451                transform: Transform::Bucket(4),
1452                direction: SortDirection::Descending,
1453                null_order: NullOrder::First,
1454            })
1455            .build(&schema)
1456            .unwrap()
1457    }
1458
1459    fn partition_spec() -> UnboundPartitionSpec {
1460        UnboundPartitionSpec::builder()
1461            .with_spec_id(0)
1462            .add_partition_field(
1463                UnboundPartitionField::builder()
1464                    .source_ids(vec![2])
1465                    .name("y")
1466                    .transform(Transform::Identity)
1467                    .build()
1468                    .unwrap(),
1469            )
1470            .unwrap()
1471            .build()
1472    }
1473
1474    fn builder_without_changes(format_version: FormatVersion) -> TableMetadataBuilder {
1475        TableMetadataBuilder::new(
1476            schema(),
1477            partition_spec(),
1478            sort_order(),
1479            TEST_LOCATION.to_string(),
1480            format_version,
1481            HashMap::new(),
1482        )
1483        .unwrap()
1484        .build()
1485        .unwrap()
1486        .metadata
1487        .into_builder(Some(
1488            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1489        ))
1490    }
1491
1492    #[test]
1493    fn test_minimal_build() {
1494        let metadata = TableMetadataBuilder::new(
1495            schema(),
1496            partition_spec(),
1497            sort_order(),
1498            TEST_LOCATION.to_string(),
1499            FormatVersion::V1,
1500            HashMap::new(),
1501        )
1502        .unwrap()
1503        .build()
1504        .unwrap()
1505        .metadata;
1506
1507        assert_eq!(metadata.format_version, FormatVersion::V1);
1508        assert_eq!(metadata.location, TEST_LOCATION);
1509        assert_eq!(metadata.current_schema_id, 0);
1510        assert_eq!(metadata.default_spec.spec_id(), 0);
1511        assert_eq!(metadata.default_sort_order_id, 1);
1512        assert_eq!(metadata.last_partition_id, 1000);
1513        assert_eq!(metadata.last_column_id, 3);
1514        assert_eq!(metadata.snapshots.len(), 0);
1515        assert_eq!(metadata.current_snapshot_id, None);
1516        assert_eq!(metadata.refs.len(), 0);
1517        assert_eq!(metadata.properties.len(), 0);
1518        assert_eq!(metadata.metadata_log.len(), 0);
1519        assert_eq!(metadata.last_sequence_number, 0);
1520        assert_eq!(metadata.last_column_id, LAST_ASSIGNED_COLUMN_ID);
1521
1522        // Test can serialize v1
1523        let _ = serde_json::to_string(&metadata).unwrap();
1524
1525        // Test can serialize v2
1526        let metadata = metadata
1527            .into_builder(Some(
1528                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1529            ))
1530            .upgrade_format_version(FormatVersion::V2)
1531            .unwrap()
1532            .build()
1533            .unwrap()
1534            .metadata;
1535
1536        assert_eq!(metadata.format_version, FormatVersion::V2);
1537        let _ = serde_json::to_string(&metadata).unwrap();
1538    }
1539
1540    #[test]
1541    fn test_build_unpartitioned_unsorted() {
1542        let schema = Schema::builder().build().unwrap();
1543        let metadata = TableMetadataBuilder::new(
1544            schema.clone(),
1545            PartitionSpec::unpartition_spec(),
1546            SortOrder::unsorted_order(),
1547            TEST_LOCATION.to_string(),
1548            FormatVersion::V2,
1549            HashMap::new(),
1550        )
1551        .unwrap()
1552        .build()
1553        .unwrap()
1554        .metadata;
1555
1556        assert_eq!(metadata.format_version, FormatVersion::V2);
1557        assert_eq!(metadata.location, TEST_LOCATION);
1558        assert_eq!(metadata.current_schema_id, 0);
1559        assert_eq!(metadata.default_spec.spec_id(), 0);
1560        assert_eq!(metadata.default_sort_order_id, 0);
1561        assert_eq!(metadata.last_partition_id, UNPARTITIONED_LAST_ASSIGNED_ID);
1562        assert_eq!(metadata.last_column_id, 0);
1563        assert_eq!(metadata.snapshots.len(), 0);
1564        assert_eq!(metadata.current_snapshot_id, None);
1565        assert_eq!(metadata.refs.len(), 0);
1566        assert_eq!(metadata.properties.len(), 0);
1567        assert_eq!(metadata.metadata_log.len(), 0);
1568        assert_eq!(metadata.last_sequence_number, 0);
1569    }
1570
1571    #[test]
1572    fn test_reassigns_ids() {
1573        let schema = Schema::builder()
1574            .with_schema_id(10)
1575            .with_fields(vec![
1576                NestedField::required(11, "a", Type::Primitive(PrimitiveType::Long)).into(),
1577                NestedField::required(12, "b", Type::Primitive(PrimitiveType::Long)).into(),
1578                NestedField::required(
1579                    13,
1580                    "struct",
1581                    Type::Struct(StructType::new(vec![
1582                        NestedField::required(14, "nested", Type::Primitive(PrimitiveType::Long))
1583                            .into(),
1584                    ])),
1585                )
1586                .into(),
1587                NestedField::required(15, "c", Type::Primitive(PrimitiveType::Long)).into(),
1588            ])
1589            .build()
1590            .unwrap();
1591        let spec = PartitionSpec::builder(schema.clone())
1592            .with_spec_id(20)
1593            .add_partition_field("a", "a", Transform::Identity)
1594            .unwrap()
1595            .add_partition_field("struct.nested", "nested_partition", Transform::Identity)
1596            .unwrap()
1597            .build()
1598            .unwrap();
1599        let sort_order = SortOrder::builder()
1600            .with_fields(vec![SortField {
1601                source_id: 11,
1602                transform: Transform::Identity,
1603                direction: SortDirection::Ascending,
1604                null_order: NullOrder::First,
1605            }])
1606            .with_order_id(10)
1607            .build(&schema)
1608            .unwrap();
1609
1610        let (fresh_schema, fresh_spec, fresh_sort_order) =
1611            TableMetadataBuilder::reassign_ids(schema, spec.into_unbound(), sort_order).unwrap();
1612
1613        let expected_schema = Schema::builder()
1614            .with_fields(vec![
1615                NestedField::required(1, "a", Type::Primitive(PrimitiveType::Long)).into(),
1616                NestedField::required(2, "b", Type::Primitive(PrimitiveType::Long)).into(),
1617                NestedField::required(
1618                    3,
1619                    "struct",
1620                    Type::Struct(StructType::new(vec![
1621                        NestedField::required(5, "nested", Type::Primitive(PrimitiveType::Long))
1622                            .into(),
1623                    ])),
1624                )
1625                .into(),
1626                NestedField::required(4, "c", Type::Primitive(PrimitiveType::Long)).into(),
1627            ])
1628            .build()
1629            .unwrap();
1630
1631        let expected_spec = PartitionSpec::builder(expected_schema.clone())
1632            .with_spec_id(0)
1633            .add_partition_field("a", "a", Transform::Identity)
1634            .unwrap()
1635            .add_partition_field("struct.nested", "nested_partition", Transform::Identity)
1636            .unwrap()
1637            .build()
1638            .unwrap();
1639
1640        let expected_sort_order = SortOrder::builder()
1641            .with_fields(vec![SortField {
1642                source_id: 1,
1643                transform: Transform::Identity,
1644                direction: SortDirection::Ascending,
1645                null_order: NullOrder::First,
1646            }])
1647            .with_order_id(1)
1648            .build(&expected_schema)
1649            .unwrap();
1650
1651        assert_eq!(fresh_schema, expected_schema);
1652        assert_eq!(fresh_spec, expected_spec);
1653        assert_eq!(fresh_sort_order, expected_sort_order);
1654    }
1655
1656    #[test]
1657    fn test_ids_are_reassigned_for_new_metadata() {
1658        let schema = schema().into_builder().with_schema_id(10).build().unwrap();
1659
1660        let metadata = TableMetadataBuilder::new(
1661            schema,
1662            partition_spec(),
1663            sort_order(),
1664            TEST_LOCATION.to_string(),
1665            FormatVersion::V1,
1666            HashMap::new(),
1667        )
1668        .unwrap()
1669        .build()
1670        .unwrap()
1671        .metadata;
1672
1673        assert_eq!(metadata.current_schema_id, 0);
1674        assert_eq!(metadata.current_schema().schema_id(), 0);
1675    }
1676
1677    #[test]
1678    fn test_new_metadata_changes() {
1679        let changes = TableMetadataBuilder::new(
1680            schema(),
1681            partition_spec(),
1682            sort_order(),
1683            TEST_LOCATION.to_string(),
1684            FormatVersion::V1,
1685            HashMap::from_iter(vec![("property 1".to_string(), "value 1".to_string())]),
1686        )
1687        .unwrap()
1688        .build()
1689        .unwrap()
1690        .changes;
1691
1692        pretty_assertions::assert_eq!(changes, vec![
1693            TableUpdate::SetLocation {
1694                location: TEST_LOCATION.to_string()
1695            },
1696            TableUpdate::AddSchema { schema: schema() },
1697            TableUpdate::SetCurrentSchema { schema_id: -1 },
1698            TableUpdate::AddSpec {
1699                // Because this is a new tables, field-ids are assigned
1700                // partition_spec() has None set for field-id
1701                spec: PartitionSpec::builder(schema())
1702                    .with_spec_id(0)
1703                    .add_unbound_field(
1704                        UnboundPartitionField::builder()
1705                            .source_ids(vec![2])
1706                            .field_id(1000)
1707                            .name("y".to_string())
1708                            .transform(Transform::Identity)
1709                            .build()
1710                            .unwrap()
1711                    )
1712                    .unwrap()
1713                    .build()
1714                    .unwrap()
1715                    .into_unbound(),
1716            },
1717            TableUpdate::SetDefaultSpec { spec_id: -1 },
1718            TableUpdate::AddSortOrder {
1719                sort_order: sort_order(),
1720            },
1721            TableUpdate::SetDefaultSortOrder { sort_order_id: -1 },
1722            TableUpdate::SetProperties {
1723                updates: HashMap::from_iter(vec![(
1724                    "property 1".to_string(),
1725                    "value 1".to_string()
1726                )]),
1727            }
1728        ]);
1729    }
1730
1731    #[test]
1732    fn test_new_metadata_changes_unpartitioned_unsorted() {
1733        let schema = Schema::builder().build().unwrap();
1734        let changes = TableMetadataBuilder::new(
1735            schema.clone(),
1736            PartitionSpec::unpartition_spec().into_unbound(),
1737            SortOrder::unsorted_order(),
1738            TEST_LOCATION.to_string(),
1739            FormatVersion::V1,
1740            HashMap::new(),
1741        )
1742        .unwrap()
1743        .build()
1744        .unwrap()
1745        .changes;
1746
1747        pretty_assertions::assert_eq!(changes, vec![
1748            TableUpdate::SetLocation {
1749                location: TEST_LOCATION.to_string()
1750            },
1751            TableUpdate::AddSchema {
1752                schema: Schema::builder().build().unwrap(),
1753            },
1754            TableUpdate::SetCurrentSchema { schema_id: -1 },
1755            TableUpdate::AddSpec {
1756                // Because this is a new tables, field-ids are assigned
1757                // partition_spec() has None set for field-id
1758                spec: PartitionSpec::builder(schema)
1759                    .with_spec_id(0)
1760                    .build()
1761                    .unwrap()
1762                    .into_unbound(),
1763            },
1764            TableUpdate::SetDefaultSpec { spec_id: -1 },
1765            TableUpdate::AddSortOrder {
1766                sort_order: SortOrder::unsorted_order(),
1767            },
1768            TableUpdate::SetDefaultSortOrder { sort_order_id: -1 },
1769        ]);
1770    }
1771
1772    #[test]
1773    fn test_add_partition_spec() {
1774        let builder = builder_without_changes(FormatVersion::V2);
1775
1776        let added_spec = UnboundPartitionSpec::builder()
1777            .with_spec_id(10)
1778            .add_partition_fields(vec![
1779                // The previous field - has field_id set
1780                UnboundPartitionField::builder()
1781                    .source_ids(vec![2])
1782                    .field_id(1000)
1783                    .name("y".to_string())
1784                    .transform(Transform::Identity)
1785                    .build()
1786                    .unwrap(),
1787                // A new field without field id - should still be without field id in changes
1788                UnboundPartitionField::builder()
1789                    .source_ids(vec![3])
1790                    .name("z".to_string())
1791                    .transform(Transform::Identity)
1792                    .build()
1793                    .unwrap(),
1794            ])
1795            .unwrap()
1796            .build();
1797
1798        let build_result = builder
1799            .add_partition_spec(added_spec.clone())
1800            .unwrap()
1801            .build()
1802            .unwrap();
1803
1804        // Spec id should be re-assigned
1805        let expected_change = added_spec.with_spec_id(1);
1806        let expected_spec = PartitionSpec::builder(schema())
1807            .with_spec_id(1)
1808            .add_unbound_field(
1809                UnboundPartitionField::builder()
1810                    .source_ids(vec![2])
1811                    .field_id(1000)
1812                    .name("y".to_string())
1813                    .transform(Transform::Identity)
1814                    .build()
1815                    .unwrap(),
1816            )
1817            .unwrap()
1818            .add_unbound_field(
1819                UnboundPartitionField::builder()
1820                    .source_ids(vec![3])
1821                    .field_id(1001)
1822                    .name("z".to_string())
1823                    .transform(Transform::Identity)
1824                    .build()
1825                    .unwrap(),
1826            )
1827            .unwrap()
1828            .build()
1829            .unwrap();
1830
1831        assert_eq!(build_result.changes.len(), 1);
1832        assert_eq!(
1833            build_result.metadata.partition_spec_by_id(1),
1834            Some(&Arc::new(expected_spec))
1835        );
1836        assert_eq!(build_result.metadata.default_spec.spec_id(), 0);
1837        assert_eq!(build_result.metadata.last_partition_id, 1001);
1838        pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSpec {
1839            spec: expected_change
1840        });
1841
1842        // Remove the spec
1843        let build_result = build_result
1844            .metadata
1845            .into_builder(Some(
1846                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1847            ))
1848            .remove_partition_specs(&[1])
1849            .unwrap()
1850            .build()
1851            .unwrap();
1852
1853        assert_eq!(build_result.changes.len(), 1);
1854        assert_eq!(build_result.metadata.partition_specs.len(), 1);
1855        assert!(build_result.metadata.partition_spec_by_id(1).is_none());
1856    }
1857
1858    #[test]
1859    fn test_set_default_partition_spec() {
1860        let builder = builder_without_changes(FormatVersion::V2);
1861        let schema = builder.get_current_schema().unwrap().clone();
1862        let added_spec = UnboundPartitionSpec::builder()
1863            .with_spec_id(10)
1864            .add_partition_field(
1865                UnboundPartitionField::builder()
1866                    .source_ids(vec![1])
1867                    .name("y_bucket[2]")
1868                    .transform(Transform::Bucket(2))
1869                    .build()
1870                    .unwrap(),
1871            )
1872            .unwrap()
1873            .build();
1874
1875        let build_result = builder
1876            .add_partition_spec(added_spec.clone())
1877            .unwrap()
1878            .set_default_partition_spec(-1)
1879            .unwrap()
1880            .build()
1881            .unwrap();
1882
1883        let expected_spec = PartitionSpec::builder(schema)
1884            .with_spec_id(1)
1885            .add_unbound_field(
1886                UnboundPartitionField::builder()
1887                    .source_ids(vec![1])
1888                    .field_id(1001)
1889                    .name("y_bucket[2]".to_string())
1890                    .transform(Transform::Bucket(2))
1891                    .build()
1892                    .unwrap(),
1893            )
1894            .unwrap()
1895            .build()
1896            .unwrap();
1897
1898        assert_eq!(build_result.changes.len(), 2);
1899        assert_eq!(build_result.metadata.default_spec, Arc::new(expected_spec));
1900        assert_eq!(build_result.changes, vec![
1901            TableUpdate::AddSpec {
1902                // Should contain the actual ID that was used
1903                spec: added_spec.with_spec_id(1)
1904            },
1905            TableUpdate::SetDefaultSpec { spec_id: -1 }
1906        ]);
1907    }
1908
1909    #[test]
1910    fn test_set_existing_default_partition_spec() {
1911        let builder = builder_without_changes(FormatVersion::V2);
1912        // Add and set an unbound spec as current
1913        let unbound_spec = UnboundPartitionSpec::builder().with_spec_id(1).build();
1914        let build_result = builder
1915            .add_partition_spec(unbound_spec.clone())
1916            .unwrap()
1917            .set_default_partition_spec(-1)
1918            .unwrap()
1919            .build()
1920            .unwrap();
1921
1922        assert_eq!(build_result.changes.len(), 2);
1923        assert_eq!(build_result.changes[0], TableUpdate::AddSpec {
1924            spec: unbound_spec.clone()
1925        });
1926        assert_eq!(build_result.changes[1], TableUpdate::SetDefaultSpec {
1927            spec_id: -1
1928        });
1929        assert_eq!(
1930            build_result.metadata.default_spec,
1931            Arc::new(
1932                unbound_spec
1933                    .bind(build_result.metadata.current_schema().clone())
1934                    .unwrap()
1935            )
1936        );
1937
1938        // Set old spec again
1939        let build_result = build_result
1940            .metadata
1941            .into_builder(Some(
1942                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1943            ))
1944            .set_default_partition_spec(0)
1945            .unwrap()
1946            .build()
1947            .unwrap();
1948
1949        assert_eq!(build_result.changes.len(), 1);
1950        assert_eq!(build_result.changes[0], TableUpdate::SetDefaultSpec {
1951            spec_id: 0
1952        });
1953        assert_eq!(
1954            build_result.metadata.default_spec,
1955            Arc::new(
1956                partition_spec()
1957                    .bind(build_result.metadata.current_schema().clone())
1958                    .unwrap()
1959            )
1960        );
1961    }
1962
1963    #[test]
1964    fn test_add_sort_order() {
1965        let builder = builder_without_changes(FormatVersion::V2);
1966
1967        let added_sort_order = SortOrder::builder()
1968            .with_order_id(10)
1969            .with_fields(vec![SortField {
1970                source_id: 1,
1971                transform: Transform::Identity,
1972                direction: SortDirection::Ascending,
1973                null_order: NullOrder::First,
1974            }])
1975            .build(&schema())
1976            .unwrap();
1977
1978        let build_result = builder
1979            .add_sort_order(added_sort_order.clone())
1980            .unwrap()
1981            .build()
1982            .unwrap();
1983
1984        let expected_sort_order = added_sort_order.with_order_id(2);
1985
1986        assert_eq!(build_result.changes.len(), 1);
1987        assert_eq!(build_result.metadata.sort_orders.keys().max(), Some(&2));
1988        pretty_assertions::assert_eq!(
1989            build_result.metadata.sort_order_by_id(2),
1990            Some(&Arc::new(expected_sort_order.clone()))
1991        );
1992        pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSortOrder {
1993            sort_order: expected_sort_order
1994        });
1995    }
1996
1997    #[test]
1998    fn test_add_compatible_schema() {
1999        let builder = builder_without_changes(FormatVersion::V2);
2000
2001        let added_schema = Schema::builder()
2002            .with_schema_id(1)
2003            .with_fields(vec![
2004                NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
2005                NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
2006                NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
2007                NestedField::required(4, "a", Type::Primitive(PrimitiveType::Long)).into(),
2008            ])
2009            .build()
2010            .unwrap();
2011
2012        let build_result = builder
2013            .add_current_schema(added_schema.clone())
2014            .unwrap()
2015            .build()
2016            .unwrap();
2017
2018        assert_eq!(build_result.changes.len(), 2);
2019        assert_eq!(build_result.metadata.schemas.keys().max(), Some(&1));
2020        pretty_assertions::assert_eq!(
2021            build_result.metadata.schema_by_id(1),
2022            Some(&Arc::new(added_schema.clone()))
2023        );
2024        pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSchema {
2025            schema: added_schema
2026        });
2027        assert_eq!(build_result.changes[1], TableUpdate::SetCurrentSchema {
2028            schema_id: -1
2029        });
2030    }
2031
2032    #[test]
2033    fn test_set_current_schema_change_is_minus_one_if_schema_was_added_in_this_change() {
2034        let builder = builder_without_changes(FormatVersion::V2);
2035
2036        let added_schema = Schema::builder()
2037            .with_schema_id(1)
2038            .with_fields(vec![
2039                NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
2040                NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
2041                NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
2042                NestedField::required(4, "a", Type::Primitive(PrimitiveType::Long)).into(),
2043            ])
2044            .build()
2045            .unwrap();
2046
2047        let build_result = builder
2048            .add_schema(added_schema.clone())
2049            .unwrap()
2050            .set_current_schema(1)
2051            .unwrap()
2052            .build()
2053            .unwrap();
2054
2055        assert_eq!(build_result.changes.len(), 2);
2056        assert_eq!(build_result.changes[1], TableUpdate::SetCurrentSchema {
2057            schema_id: -1
2058        });
2059    }
2060
2061    #[test]
2062    fn test_no_metadata_log_for_create_table() {
2063        let build_result = TableMetadataBuilder::new(
2064            schema(),
2065            partition_spec(),
2066            sort_order(),
2067            TEST_LOCATION.to_string(),
2068            FormatVersion::V2,
2069            HashMap::new(),
2070        )
2071        .unwrap()
2072        .build()
2073        .unwrap();
2074
2075        assert_eq!(build_result.metadata.metadata_log.len(), 0);
2076    }
2077
2078    #[test]
2079    fn test_table_properties_view_reflects_metadata_updates() {
2080        let property = TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string();
2081        let metadata = builder_without_changes(FormatVersion::V2)
2082            .set_properties(HashMap::from([(property.clone(), "7".to_string())]))
2083            .unwrap()
2084            .build()
2085            .unwrap()
2086            .metadata;
2087
2088        assert_eq!(metadata.table_properties().commit_num_retries().unwrap(), 7);
2089
2090        let metadata = metadata
2091            .into_builder(None)
2092            .remove_properties(&[property])
2093            .unwrap()
2094            .build()
2095            .unwrap()
2096            .metadata;
2097
2098        assert_eq!(
2099            metadata.table_properties().commit_num_retries().unwrap(),
2100            TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
2101        );
2102    }
2103
2104    #[test]
2105    fn test_no_metadata_log_entry_for_no_previous_location() {
2106        // Used for first commit after stage-creation of tables
2107        let metadata = builder_without_changes(FormatVersion::V2)
2108            .build()
2109            .unwrap()
2110            .metadata;
2111        assert_eq!(metadata.metadata_log.len(), 1);
2112
2113        let build_result = metadata
2114            .into_builder(None)
2115            .set_properties(HashMap::from_iter(vec![(
2116                "foo".to_string(),
2117                "bar".to_string(),
2118            )]))
2119            .unwrap()
2120            .build()
2121            .unwrap();
2122
2123        assert_eq!(build_result.metadata.metadata_log.len(), 1);
2124    }
2125
2126    #[test]
2127    fn test_from_metadata_generates_metadata_log() {
2128        let metadata_path = "s3://bucket/test/location/metadata/metadata1.json";
2129        let builder = TableMetadataBuilder::new(
2130            schema(),
2131            partition_spec(),
2132            sort_order(),
2133            TEST_LOCATION.to_string(),
2134            FormatVersion::V2,
2135            HashMap::new(),
2136        )
2137        .unwrap()
2138        .build()
2139        .unwrap()
2140        .metadata
2141        .into_builder(Some(metadata_path.to_string()));
2142
2143        let builder = builder
2144            .add_default_sort_order(SortOrder::unsorted_order())
2145            .unwrap();
2146
2147        let build_result = builder.build().unwrap();
2148
2149        assert_eq!(build_result.metadata.metadata_log.len(), 1);
2150        assert_eq!(
2151            build_result.metadata.metadata_log[0].metadata_file,
2152            metadata_path
2153        );
2154    }
2155
2156    #[test]
2157    fn test_set_ref() {
2158        let builder = builder_without_changes(FormatVersion::V2);
2159
2160        let snapshot = Snapshot::builder()
2161            .with_snapshot_id(1)
2162            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2163            .with_sequence_number(0)
2164            .with_schema_id(0)
2165            .with_manifest_list("/snap-1.avro")
2166            .with_summary(Summary {
2167                operation: Operation::Append,
2168                additional_properties: HashMap::from_iter(vec![
2169                    (
2170                        "spark.app.id".to_string(),
2171                        "local-1662532784305".to_string(),
2172                    ),
2173                    ("added-data-files".to_string(), "4".to_string()),
2174                    ("added-records".to_string(), "4".to_string()),
2175                    ("added-files-size".to_string(), "6001".to_string()),
2176                ]),
2177            })
2178            .build();
2179
2180        let builder = builder.add_snapshot(snapshot.clone()).unwrap();
2181
2182        assert!(
2183            builder
2184                .clone()
2185                .set_ref(MAIN_BRANCH, SnapshotReference {
2186                    snapshot_id: 10,
2187                    retention: SnapshotRetention::Branch {
2188                        min_snapshots_to_keep: Some(10),
2189                        max_snapshot_age_ms: None,
2190                        max_ref_age_ms: None,
2191                    },
2192                })
2193                .unwrap_err()
2194                .to_string()
2195                .contains("Cannot set 'main' to unknown snapshot: '10'")
2196        );
2197
2198        let build_result = builder
2199            .set_ref(MAIN_BRANCH, SnapshotReference {
2200                snapshot_id: 1,
2201                retention: SnapshotRetention::Branch {
2202                    min_snapshots_to_keep: Some(10),
2203                    max_snapshot_age_ms: None,
2204                    max_ref_age_ms: None,
2205                },
2206            })
2207            .unwrap()
2208            .build()
2209            .unwrap();
2210        assert_eq!(build_result.metadata.snapshots.len(), 1);
2211        assert_eq!(
2212            build_result.metadata.snapshot_by_id(1),
2213            Some(&Arc::new(snapshot.clone()))
2214        );
2215        assert_eq!(build_result.metadata.snapshot_log, vec![SnapshotLog {
2216            snapshot_id: 1,
2217            timestamp_ms: snapshot.timestamp_ms()
2218        }])
2219    }
2220
2221    #[test]
2222    fn test_snapshot_log_skips_intermediates() {
2223        let builder = builder_without_changes(FormatVersion::V2);
2224
2225        let snapshot_1 = Snapshot::builder()
2226            .with_snapshot_id(1)
2227            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2228            .with_sequence_number(0)
2229            .with_schema_id(0)
2230            .with_manifest_list("/snap-1.avro")
2231            .with_summary(Summary {
2232                operation: Operation::Append,
2233                additional_properties: HashMap::from_iter(vec![
2234                    (
2235                        "spark.app.id".to_string(),
2236                        "local-1662532784305".to_string(),
2237                    ),
2238                    ("added-data-files".to_string(), "4".to_string()),
2239                    ("added-records".to_string(), "4".to_string()),
2240                    ("added-files-size".to_string(), "6001".to_string()),
2241                ]),
2242            })
2243            .build();
2244
2245        let snapshot_2 = Snapshot::builder()
2246            .with_snapshot_id(2)
2247            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2248            .with_sequence_number(0)
2249            .with_schema_id(0)
2250            .with_manifest_list("/snap-1.avro")
2251            .with_summary(Summary {
2252                operation: Operation::Append,
2253                additional_properties: HashMap::from_iter(vec![
2254                    (
2255                        "spark.app.id".to_string(),
2256                        "local-1662532784305".to_string(),
2257                    ),
2258                    ("added-data-files".to_string(), "4".to_string()),
2259                    ("added-records".to_string(), "4".to_string()),
2260                    ("added-files-size".to_string(), "6001".to_string()),
2261                ]),
2262            })
2263            .build();
2264
2265        let result = builder
2266            .add_snapshot(snapshot_1)
2267            .unwrap()
2268            .set_ref(MAIN_BRANCH, SnapshotReference {
2269                snapshot_id: 1,
2270                retention: SnapshotRetention::Branch {
2271                    min_snapshots_to_keep: Some(10),
2272                    max_snapshot_age_ms: None,
2273                    max_ref_age_ms: None,
2274                },
2275            })
2276            .unwrap()
2277            .set_branch_snapshot(snapshot_2.clone(), MAIN_BRANCH)
2278            .unwrap()
2279            .build()
2280            .unwrap();
2281
2282        assert_eq!(result.metadata.snapshot_log, vec![SnapshotLog {
2283            snapshot_id: 2,
2284            timestamp_ms: snapshot_2.timestamp_ms()
2285        }]);
2286        assert_eq!(result.metadata.current_snapshot().unwrap().snapshot_id(), 2);
2287    }
2288
2289    #[test]
2290    fn test_remove_main_ref_keeps_snapshot_log() {
2291        let builder = builder_without_changes(FormatVersion::V2);
2292
2293        let snapshot = Snapshot::builder()
2294            .with_snapshot_id(1)
2295            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2296            .with_sequence_number(0)
2297            .with_schema_id(0)
2298            .with_manifest_list("/snap-1.avro")
2299            .with_summary(Summary {
2300                operation: Operation::Append,
2301                additional_properties: HashMap::from_iter(vec![
2302                    (
2303                        "spark.app.id".to_string(),
2304                        "local-1662532784305".to_string(),
2305                    ),
2306                    ("added-data-files".to_string(), "4".to_string()),
2307                    ("added-records".to_string(), "4".to_string()),
2308                    ("added-files-size".to_string(), "6001".to_string()),
2309                ]),
2310            })
2311            .build();
2312
2313        let result = builder
2314            .add_snapshot(snapshot.clone())
2315            .unwrap()
2316            .set_ref(MAIN_BRANCH, SnapshotReference {
2317                snapshot_id: 1,
2318                retention: SnapshotRetention::Branch {
2319                    min_snapshots_to_keep: Some(10),
2320                    max_snapshot_age_ms: None,
2321                    max_ref_age_ms: None,
2322                },
2323            })
2324            .unwrap()
2325            .build()
2326            .unwrap();
2327
2328        // Verify snapshot log was created
2329        assert_eq!(result.metadata.snapshot_log.len(), 1);
2330        assert_eq!(result.metadata.snapshot_log[0].snapshot_id, 1);
2331        assert_eq!(result.metadata.current_snapshot_id, Some(1));
2332
2333        // Remove the main ref
2334        let result_after_remove = result
2335            .metadata
2336            .into_builder(Some(
2337                "s3://bucket/test/location/metadata/metadata2.json".to_string(),
2338            ))
2339            .remove_ref(MAIN_BRANCH)
2340            .build()
2341            .unwrap();
2342
2343        // Verify snapshot log is kept even after removing main ref
2344        assert_eq!(result_after_remove.metadata.snapshot_log.len(), 1);
2345        assert_eq!(result_after_remove.metadata.snapshot_log[0].snapshot_id, 1);
2346        assert_eq!(result_after_remove.metadata.current_snapshot_id, None);
2347        assert_eq!(result_after_remove.changes.len(), 1);
2348        assert_eq!(
2349            result_after_remove.changes[0],
2350            TableUpdate::RemoveSnapshotRef {
2351                ref_name: MAIN_BRANCH.to_string()
2352            }
2353        );
2354    }
2355
2356    #[test]
2357    fn test_set_branch_snapshot_creates_branch_if_not_exists() {
2358        let builder = builder_without_changes(FormatVersion::V2);
2359
2360        let snapshot = Snapshot::builder()
2361            .with_snapshot_id(2)
2362            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2363            .with_sequence_number(0)
2364            .with_schema_id(0)
2365            .with_manifest_list("/snap-1.avro")
2366            .with_summary(Summary {
2367                operation: Operation::Append,
2368                additional_properties: HashMap::new(),
2369            })
2370            .build();
2371
2372        let build_result = builder
2373            .set_branch_snapshot(snapshot.clone(), "new_branch")
2374            .unwrap()
2375            .build()
2376            .unwrap();
2377
2378        let reference = SnapshotReference {
2379            snapshot_id: 2,
2380            retention: SnapshotRetention::Branch {
2381                min_snapshots_to_keep: None,
2382                max_snapshot_age_ms: None,
2383                max_ref_age_ms: None,
2384            },
2385        };
2386
2387        assert_eq!(build_result.metadata.refs.len(), 1);
2388        assert_eq!(
2389            build_result.metadata.refs.get("new_branch"),
2390            Some(&reference)
2391        );
2392        assert_eq!(build_result.changes, vec![
2393            TableUpdate::AddSnapshot { snapshot },
2394            TableUpdate::SetSnapshotRef {
2395                ref_name: "new_branch".to_string(),
2396                reference
2397            }
2398        ]);
2399    }
2400
2401    #[test]
2402    fn test_cannot_add_duplicate_snapshot_id() {
2403        let builder = builder_without_changes(FormatVersion::V2);
2404
2405        let snapshot = Snapshot::builder()
2406            .with_snapshot_id(2)
2407            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2408            .with_sequence_number(0)
2409            .with_schema_id(0)
2410            .with_manifest_list("/snap-1.avro")
2411            .with_summary(Summary {
2412                operation: Operation::Append,
2413                additional_properties: HashMap::from_iter(vec![
2414                    (
2415                        "spark.app.id".to_string(),
2416                        "local-1662532784305".to_string(),
2417                    ),
2418                    ("added-data-files".to_string(), "4".to_string()),
2419                    ("added-records".to_string(), "4".to_string()),
2420                    ("added-files-size".to_string(), "6001".to_string()),
2421                ]),
2422            })
2423            .build();
2424
2425        let builder = builder.add_snapshot(snapshot.clone()).unwrap();
2426        builder.add_snapshot(snapshot).unwrap_err();
2427    }
2428
2429    #[test]
2430    fn test_add_incompatible_current_schema_fails() {
2431        let builder = builder_without_changes(FormatVersion::V2);
2432
2433        let added_schema = Schema::builder()
2434            .with_schema_id(1)
2435            .with_fields(vec![])
2436            .build()
2437            .unwrap();
2438
2439        let err = builder
2440            .add_current_schema(added_schema)
2441            .unwrap()
2442            .build()
2443            .unwrap_err();
2444
2445        assert!(
2446            err.to_string()
2447                .contains("Cannot find partition source field")
2448        );
2449    }
2450
2451    #[test]
2452    fn test_add_partition_spec_for_v1_requires_sequential_ids() {
2453        let builder = builder_without_changes(FormatVersion::V1);
2454
2455        let added_spec = UnboundPartitionSpec::builder()
2456            .with_spec_id(10)
2457            .add_partition_fields(vec![
2458                UnboundPartitionField::builder()
2459                    .source_ids(vec![2])
2460                    .field_id(1000)
2461                    .name("y".to_string())
2462                    .transform(Transform::Identity)
2463                    .build()
2464                    .unwrap(),
2465                UnboundPartitionField::builder()
2466                    .source_ids(vec![3])
2467                    .field_id(1002)
2468                    .name("z".to_string())
2469                    .transform(Transform::Identity)
2470                    .build()
2471                    .unwrap(),
2472            ])
2473            .unwrap()
2474            .build();
2475
2476        let err = builder.add_partition_spec(added_spec).unwrap_err();
2477        assert!(err.to_string().contains(
2478            "Cannot add partition spec with non-sequential field ids to format version 1 table"
2479        ));
2480    }
2481
2482    #[test]
2483    fn test_add_partition_spec_accepts_void_fields_of_removed_fields() {
2484        let builder = builder_without_changes(FormatVersion::V1);
2485        let void_field = |field_id: i32, name: &str| {
2486            UnboundPartitionField::builder()
2487                .source_ids(vec![2])
2488                .field_id(field_id)
2489                .name(name)
2490                .transform(Transform::Void)
2491                .build()
2492                .unwrap()
2493        };
2494        let added_spec = UnboundPartitionSpec {
2495            spec_id: None,
2496            fields: vec![void_field(1000, "y"), void_field(1001, "y_null")],
2497        };
2498
2499        let metadata = builder
2500            .add_default_partition_spec(added_spec.clone())
2501            .unwrap()
2502            .build()
2503            .unwrap()
2504            .metadata;
2505
2506        let expected = added_spec.with_spec_id(1).bind(schema()).unwrap();
2507        assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
2508        assert_eq!(metadata.last_partition_id, 1001);
2509    }
2510
2511    #[test]
2512    fn test_add_partition_spec_accepts_two_time_granularities_of_one_column() {
2513        let schema = Schema::builder()
2514            .with_fields(vec![
2515                NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(),
2516            ])
2517            .build()
2518            .unwrap();
2519        let time_field = |field_id: i32, name: &str, transform: Transform| {
2520            UnboundPartitionField::builder()
2521                .source_ids(vec![1])
2522                .field_id(field_id)
2523                .name(name)
2524                .transform(transform)
2525                .build()
2526                .unwrap()
2527        };
2528        let added_spec = UnboundPartitionSpec {
2529            spec_id: None,
2530            fields: vec![
2531                time_field(1000, "ts_day", Transform::Day),
2532                time_field(1001, "ts_hour", Transform::Hour),
2533            ],
2534        };
2535        let metadata = TableMetadataBuilder::new(
2536            schema.clone(),
2537            PartitionSpec::unpartition_spec(),
2538            SortOrder::unsorted_order(),
2539            TEST_LOCATION.to_string(),
2540            FormatVersion::V2,
2541            HashMap::new(),
2542        )
2543        .unwrap()
2544        .build()
2545        .unwrap()
2546        .metadata;
2547
2548        let metadata = metadata
2549            .into_builder(None)
2550            .add_default_partition_spec(added_spec.clone())
2551            .unwrap()
2552            .build()
2553            .unwrap()
2554            .metadata;
2555        // Every later commit re-validates the default spec.
2556        let metadata = metadata
2557            .into_builder(None)
2558            .set_properties(HashMap::from([("key".to_string(), "value".to_string())]))
2559            .unwrap()
2560            .build()
2561            .unwrap()
2562            .metadata;
2563
2564        let expected = added_spec.with_spec_id(1).bind(schema).unwrap();
2565        assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
2566    }
2567
2568    #[test]
2569    fn test_expire_metadata_log() {
2570        let builder = builder_without_changes(FormatVersion::V2);
2571        let metadata = builder
2572            .set_properties(HashMap::from_iter(vec![(
2573                TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX.to_string(),
2574                "2".to_string(),
2575            )]))
2576            .unwrap()
2577            .build()
2578            .unwrap();
2579        assert_eq!(metadata.metadata.metadata_log.len(), 1);
2580        assert_eq!(metadata.expired_metadata_logs.len(), 0);
2581
2582        let metadata = metadata
2583            .metadata
2584            .into_builder(Some("path2".to_string()))
2585            .set_properties(HashMap::from_iter(vec![(
2586                "change_nr".to_string(),
2587                "1".to_string(),
2588            )]))
2589            .unwrap()
2590            .build()
2591            .unwrap();
2592
2593        assert_eq!(metadata.metadata.metadata_log.len(), 2);
2594        assert_eq!(metadata.expired_metadata_logs.len(), 0);
2595
2596        let metadata = metadata
2597            .metadata
2598            .into_builder(Some("path2".to_string()))
2599            .set_properties(HashMap::from_iter(vec![(
2600                "change_nr".to_string(),
2601                "2".to_string(),
2602            )]))
2603            .unwrap()
2604            .build()
2605            .unwrap();
2606        assert_eq!(metadata.metadata.metadata_log.len(), 2);
2607        assert_eq!(metadata.expired_metadata_logs.len(), 1);
2608    }
2609
2610    #[test]
2611    fn test_v2_sequence_number_cannot_decrease() {
2612        let builder = builder_without_changes(FormatVersion::V2);
2613
2614        let snapshot = Snapshot::builder()
2615            .with_snapshot_id(1)
2616            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2617            .with_sequence_number(1)
2618            .with_schema_id(0)
2619            .with_manifest_list("/snap-1")
2620            .with_summary(Summary {
2621                operation: Operation::Append,
2622                additional_properties: HashMap::new(),
2623            })
2624            .build();
2625
2626        let builder = builder
2627            .add_snapshot(snapshot.clone())
2628            .unwrap()
2629            .set_ref(MAIN_BRANCH, SnapshotReference {
2630                snapshot_id: 1,
2631                retention: SnapshotRetention::Branch {
2632                    min_snapshots_to_keep: Some(10),
2633                    max_snapshot_age_ms: None,
2634                    max_ref_age_ms: None,
2635                },
2636            })
2637            .unwrap();
2638
2639        let snapshot = Snapshot::builder()
2640            .with_snapshot_id(2)
2641            .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2642            .with_sequence_number(0)
2643            .with_schema_id(0)
2644            .with_manifest_list("/snap-0")
2645            .with_parent_snapshot_id(Some(1))
2646            .with_summary(Summary {
2647                operation: Operation::Append,
2648                additional_properties: HashMap::new(),
2649            })
2650            .build();
2651
2652        let err = builder
2653            .set_branch_snapshot(snapshot, MAIN_BRANCH)
2654            .unwrap_err();
2655        assert!(
2656            err.to_string()
2657                .contains("Cannot add snapshot with sequence number")
2658        );
2659    }
2660
2661    #[test]
2662    fn test_default_spec_cannot_be_removed() {
2663        let builder = builder_without_changes(FormatVersion::V2);
2664
2665        builder.remove_partition_specs(&[0]).unwrap_err();
2666    }
2667
2668    #[test]
2669    fn test_statistics() {
2670        let builder = builder_without_changes(FormatVersion::V2);
2671
2672        let statistics = StatisticsFile {
2673            snapshot_id: 3055729675574597004,
2674            statistics_path: "s3://a/b/stats.puffin".to_string(),
2675            file_size_in_bytes: 413,
2676            file_footer_size_in_bytes: 42,
2677            key_metadata: None,
2678            blob_metadata: vec![BlobMetadata {
2679                snapshot_id: 3055729675574597004,
2680                sequence_number: 1,
2681                fields: vec![1],
2682                r#type: "ndv".to_string(),
2683                properties: HashMap::new(),
2684            }],
2685        };
2686        let build_result = builder.set_statistics(statistics.clone()).build().unwrap();
2687
2688        assert_eq!(
2689            build_result.metadata.statistics,
2690            HashMap::from_iter(vec![(3055729675574597004, statistics.clone())])
2691        );
2692        assert_eq!(build_result.changes, vec![TableUpdate::SetStatistics {
2693            statistics: statistics.clone()
2694        }]);
2695
2696        // Remove
2697        let builder = build_result.metadata.into_builder(None);
2698        let build_result = builder
2699            .remove_statistics(statistics.snapshot_id)
2700            .build()
2701            .unwrap();
2702
2703        assert_eq!(build_result.metadata.statistics.len(), 0);
2704        assert_eq!(build_result.changes, vec![TableUpdate::RemoveStatistics {
2705            snapshot_id: statistics.snapshot_id
2706        }]);
2707
2708        // Remove again yields no changes
2709        let builder = build_result.metadata.into_builder(None);
2710        let build_result = builder
2711            .remove_statistics(statistics.snapshot_id)
2712            .build()
2713            .unwrap();
2714        assert_eq!(build_result.metadata.statistics.len(), 0);
2715        assert_eq!(build_result.changes.len(), 0);
2716    }
2717
2718    #[test]
2719    fn test_add_partition_statistics() {
2720        let builder = builder_without_changes(FormatVersion::V2);
2721
2722        let statistics = PartitionStatisticsFile {
2723            snapshot_id: 3055729675574597004,
2724            statistics_path: "s3://a/b/partition-stats.parquet".to_string(),
2725            file_size_in_bytes: 43,
2726        };
2727
2728        let build_result = builder
2729            .set_partition_statistics(statistics.clone())
2730            .build()
2731            .unwrap();
2732        assert_eq!(
2733            build_result.metadata.partition_statistics,
2734            HashMap::from_iter(vec![(3055729675574597004, statistics.clone())])
2735        );
2736        assert_eq!(build_result.changes, vec![
2737            TableUpdate::SetPartitionStatistics {
2738                partition_statistics: statistics.clone()
2739            }
2740        ]);
2741
2742        // Remove
2743        let builder = build_result.metadata.into_builder(None);
2744        let build_result = builder
2745            .remove_partition_statistics(statistics.snapshot_id)
2746            .build()
2747            .unwrap();
2748        assert_eq!(build_result.metadata.partition_statistics.len(), 0);
2749        assert_eq!(build_result.changes, vec![
2750            TableUpdate::RemovePartitionStatistics {
2751                snapshot_id: statistics.snapshot_id
2752            }
2753        ]);
2754
2755        // Remove again yields no changes
2756        let builder = build_result.metadata.into_builder(None);
2757        let build_result = builder
2758            .remove_partition_statistics(statistics.snapshot_id)
2759            .build()
2760            .unwrap();
2761        assert_eq!(build_result.metadata.partition_statistics.len(), 0);
2762        assert_eq!(build_result.changes.len(), 0);
2763    }
2764
2765    #[test]
2766    fn last_update_increased_for_property_only_update() {
2767        let builder = builder_without_changes(FormatVersion::V2);
2768
2769        let metadata = builder.build().unwrap().metadata;
2770        let last_updated_ms = metadata.last_updated_ms;
2771        sleep(std::time::Duration::from_millis(2));
2772
2773        let build_result = metadata
2774            .into_builder(Some(
2775                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
2776            ))
2777            .set_properties(HashMap::from_iter(vec![(
2778                "foo".to_string(),
2779                "bar".to_string(),
2780            )]))
2781            .unwrap()
2782            .build()
2783            .unwrap();
2784
2785        assert!(
2786            build_result.metadata.last_updated_ms > last_updated_ms,
2787            "{} > {}",
2788            build_result.metadata.last_updated_ms,
2789            last_updated_ms
2790        );
2791    }
2792
2793    #[test]
2794    fn test_construct_default_main_branch() {
2795        // Load the table without ref
2796        let file = File::open(format!(
2797            "{}/testdata/table_metadata/{}",
2798            env!("CARGO_MANIFEST_DIR"),
2799            "TableMetadataV2Valid.json"
2800        ))
2801        .unwrap();
2802        let reader = BufReader::new(file);
2803        let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
2804
2805        let table = Table::builder()
2806            .metadata(resp)
2807            .metadata_location("s3://bucket/test/location/metadata/v1.json")
2808            .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
2809            .file_io(FileIO::new_with_memory())
2810            .runtime(test_runtime())
2811            .build()
2812            .unwrap();
2813
2814        assert_eq!(
2815            table.metadata().refs.get(MAIN_BRANCH).unwrap().snapshot_id,
2816            table.metadata().current_snapshot_id().unwrap()
2817        );
2818    }
2819
2820    #[test]
2821    fn test_active_schema_cannot_be_removed() {
2822        let builder = builder_without_changes(FormatVersion::V2);
2823        builder.remove_schemas(&[0]).unwrap_err();
2824    }
2825
2826    #[test]
2827    fn test_remove_schemas() {
2828        let file = File::open(format!(
2829            "{}/testdata/table_metadata/{}",
2830            env!("CARGO_MANIFEST_DIR"),
2831            "TableMetadataV2Valid.json"
2832        ))
2833        .unwrap();
2834        let reader = BufReader::new(file);
2835        let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
2836
2837        let table = Table::builder()
2838            .metadata(resp)
2839            .metadata_location("s3://bucket/test/location/metadata/v1.json")
2840            .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
2841            .file_io(FileIO::new_with_memory())
2842            .runtime(test_runtime())
2843            .build()
2844            .unwrap();
2845
2846        assert_eq!(2, table.metadata().schemas.len());
2847
2848        {
2849            // can not remove active schema
2850            let meta_data_builder = table.metadata().clone().into_builder(None);
2851            meta_data_builder.remove_schemas(&[1]).unwrap_err();
2852        }
2853
2854        let mut meta_data_builder = table.metadata().clone().into_builder(None);
2855        meta_data_builder = meta_data_builder.remove_schemas(&[0]).unwrap();
2856        let build_result = meta_data_builder.build().unwrap();
2857        assert_eq!(1, build_result.metadata.schemas.len());
2858        assert_eq!(1, build_result.metadata.current_schema_id);
2859        assert_eq!(1, build_result.metadata.current_schema().schema_id());
2860        assert_eq!(1, build_result.changes.len());
2861
2862        let remove_schema_ids =
2863            if let TableUpdate::RemoveSchemas { schema_ids } = &build_result.changes[0] {
2864                schema_ids
2865            } else {
2866                unreachable!("Expected RemoveSchema change")
2867            };
2868        assert_eq!(remove_schema_ids, &[0]);
2869    }
2870
2871    #[test]
2872    fn test_schema_evolution_now_correctly_validates_partition_field_name_conflicts() {
2873        let initial_schema = Schema::builder()
2874            .with_fields(vec![
2875                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2876            ])
2877            .build()
2878            .unwrap();
2879
2880        let partition_spec_with_bucket = UnboundPartitionSpec::builder()
2881            .with_spec_id(0)
2882            .add_partition_field(
2883                UnboundPartitionField::builder()
2884                    .source_ids(vec![1])
2885                    .name("bucket_data")
2886                    .transform(Transform::Bucket(16))
2887                    .build()
2888                    .unwrap(),
2889            )
2890            .unwrap()
2891            .build();
2892
2893        let metadata = TableMetadataBuilder::new(
2894            initial_schema,
2895            partition_spec_with_bucket,
2896            SortOrder::unsorted_order(),
2897            TEST_LOCATION.to_string(),
2898            FormatVersion::V2,
2899            HashMap::new(),
2900        )
2901        .unwrap()
2902        .build()
2903        .unwrap()
2904        .metadata;
2905
2906        let partition_field_names: Vec<String> = metadata
2907            .default_partition_spec()
2908            .fields()
2909            .iter()
2910            .map(|f| f.name.clone())
2911            .collect();
2912        assert!(partition_field_names.contains(&"bucket_data".to_string()));
2913
2914        let evolved_schema = Schema::builder()
2915            .with_fields(vec![
2916                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2917                // Adding a schema field with the same name as an existing partition field
2918                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
2919            ])
2920            .build()
2921            .unwrap();
2922
2923        let builder = metadata.into_builder(Some(
2924            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
2925        ));
2926
2927        // Try to add the evolved schema - this should now fail immediately with a clear error
2928        let result = builder.add_current_schema(evolved_schema);
2929
2930        assert!(result.is_err());
2931        let error = result.unwrap_err();
2932        let error_message = error.message();
2933        assert!(error_message.contains("Cannot set current schema"));
2934        assert!(error_message.contains("sourced from field id 1"));
2935    }
2936
2937    #[test]
2938    fn test_schema_evolution_should_validate_on_schema_add_not_metadata_build() {
2939        let initial_schema = Schema::builder()
2940            .with_fields(vec![
2941                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2942            ])
2943            .build()
2944            .unwrap();
2945
2946        let partition_spec = UnboundPartitionSpec::builder()
2947            .with_spec_id(0)
2948            .add_partition_field(
2949                UnboundPartitionField::builder()
2950                    .source_ids(vec![1])
2951                    .name("partition_col")
2952                    .transform(Transform::Bucket(16))
2953                    .build()
2954                    .unwrap(),
2955            )
2956            .unwrap()
2957            .build();
2958
2959        let metadata = TableMetadataBuilder::new(
2960            initial_schema,
2961            partition_spec,
2962            SortOrder::unsorted_order(),
2963            TEST_LOCATION.to_string(),
2964            FormatVersion::V2,
2965            HashMap::new(),
2966        )
2967        .unwrap()
2968        .build()
2969        .unwrap()
2970        .metadata;
2971
2972        let non_conflicting_schema = Schema::builder()
2973            .with_fields(vec![
2974                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2975                NestedField::required(2, "new_field", Type::Primitive(PrimitiveType::Int)).into(),
2976            ])
2977            .build()
2978            .unwrap();
2979
2980        // This should succeed since there's no name conflict
2981        let result = metadata
2982            .clone()
2983            .into_builder(Some("test_location".to_string()))
2984            .add_current_schema(non_conflicting_schema)
2985            .unwrap()
2986            .build();
2987
2988        assert!(result.is_ok());
2989    }
2990
2991    #[test]
2992    fn test_partition_spec_evolution_validates_schema_field_name_conflicts() {
2993        let initial_schema = Schema::builder()
2994            .with_fields(vec![
2995                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2996                NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
2997                    .into(),
2998            ])
2999            .build()
3000            .unwrap();
3001
3002        let partition_spec = UnboundPartitionSpec::builder()
3003            .with_spec_id(0)
3004            .add_partition_field(
3005                UnboundPartitionField::builder()
3006                    .source_ids(vec![1])
3007                    .name("data_bucket")
3008                    .transform(Transform::Bucket(16))
3009                    .build()
3010                    .unwrap(),
3011            )
3012            .unwrap()
3013            .build();
3014
3015        let metadata = TableMetadataBuilder::new(
3016            initial_schema,
3017            partition_spec,
3018            SortOrder::unsorted_order(),
3019            TEST_LOCATION.to_string(),
3020            FormatVersion::V2,
3021            HashMap::new(),
3022        )
3023        .unwrap()
3024        .build()
3025        .unwrap()
3026        .metadata;
3027
3028        let builder = metadata.into_builder(Some(
3029            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3030        ));
3031
3032        let conflicting_partition_spec = UnboundPartitionSpec::builder()
3033            .with_spec_id(1)
3034            .add_partition_field(
3035                UnboundPartitionField::builder()
3036                    .source_ids(vec![1])
3037                    .name("existing_field")
3038                    .transform(Transform::Bucket(8))
3039                    .build()
3040                    .unwrap(),
3041            )
3042            .unwrap()
3043            .build();
3044
3045        let result = builder.add_partition_spec(conflicting_partition_spec);
3046
3047        assert!(result.is_err());
3048        let error = result.unwrap_err();
3049        let error_message = error.message();
3050        // The error comes from our multi-version validation
3051        assert!(error_message.contains(
3052            "Cannot create partition with name 'existing_field' that conflicts with schema field"
3053        ));
3054        assert!(error_message.contains("and is not an identity or void transform"));
3055    }
3056
3057    #[test]
3058    fn test_schema_evolution_validates_against_all_historical_schemas() {
3059        // Create a table with an initial schema that has a field "existing_field"
3060        let initial_schema = Schema::builder()
3061            .with_fields(vec![
3062                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3063                NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
3064                    .into(),
3065            ])
3066            .build()
3067            .unwrap();
3068
3069        let partition_spec = UnboundPartitionSpec::builder()
3070            .with_spec_id(0)
3071            .add_partition_field(
3072                UnboundPartitionField::builder()
3073                    .source_ids(vec![1])
3074                    .name("bucket_data")
3075                    .transform(Transform::Bucket(16))
3076                    .build()
3077                    .unwrap(),
3078            )
3079            .unwrap()
3080            .build();
3081
3082        let metadata = TableMetadataBuilder::new(
3083            initial_schema,
3084            partition_spec,
3085            SortOrder::unsorted_order(),
3086            TEST_LOCATION.to_string(),
3087            FormatVersion::V2,
3088            HashMap::new(),
3089        )
3090        .unwrap()
3091        .build()
3092        .unwrap()
3093        .metadata;
3094
3095        // Add a second schema that removes the existing_field but keeps the data field
3096        let second_schema = Schema::builder()
3097            .with_fields(vec![
3098                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3099                NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3100                    .into(),
3101            ])
3102            .build()
3103            .unwrap();
3104
3105        let metadata = metadata
3106            .into_builder(Some("test_location".to_string()))
3107            .add_current_schema(second_schema)
3108            .unwrap()
3109            .build()
3110            .unwrap()
3111            .metadata;
3112
3113        // Now try to add a third schema that reintroduces "existing_field"
3114        // This should succeed because "existing_field" exists in a historical schema,
3115        // even though there's a partition field named "bucket_data"
3116        let third_schema = Schema::builder()
3117            .with_fields(vec![
3118                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3119                NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3120                    .into(),
3121                NestedField::required(4, "existing_field", Type::Primitive(PrimitiveType::Int))
3122                    .into(),
3123            ])
3124            .build()
3125            .unwrap();
3126
3127        let builder = metadata
3128            .clone()
3129            .into_builder(Some("test_location".to_string()));
3130
3131        // This should succeed because "existing_field" exists in a historical schema
3132        let result = builder.add_current_schema(third_schema);
3133        assert!(result.is_ok());
3134
3135        // However, trying to add a schema field that conflicts with the partition field
3136        // and doesn't exist in any historical schema should fail
3137        let conflicting_schema = Schema::builder()
3138            .with_fields(vec![
3139                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3140                NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3141                    .into(),
3142                NestedField::required(4, "existing_field", Type::Primitive(PrimitiveType::Int))
3143                    .into(),
3144                NestedField::required(5, "bucket_data", Type::Primitive(PrimitiveType::String))
3145                    .into(), // conflicts with partition field
3146            ])
3147            .build()
3148            .unwrap();
3149
3150        let builder2 = metadata.into_builder(Some("test_location".to_string()));
3151        let result2 = builder2.add_current_schema(conflicting_schema);
3152
3153        // This should fail because "bucket_data" takes the name of a partition field whose
3154        // source column `data` is still in the schema
3155        assert!(result2.is_err());
3156        let error = result2.unwrap_err();
3157        assert!(error.message().contains("Cannot set current schema"));
3158        assert!(error.message().contains("sourced from field id 1"));
3159    }
3160
3161    #[test]
3162    fn test_schema_evolution_allows_existing_partition_field_if_exists_in_historical_schema() {
3163        // Create initial schema with a field
3164        let initial_schema = Schema::builder()
3165            .with_fields(vec![
3166                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3167                NestedField::required(2, "partition_data", Type::Primitive(PrimitiveType::Int))
3168                    .into(),
3169            ])
3170            .build()
3171            .unwrap();
3172
3173        let partition_spec = UnboundPartitionSpec::builder()
3174            .with_spec_id(0)
3175            .add_partition_field(
3176                UnboundPartitionField::builder()
3177                    .source_ids(vec![2])
3178                    .name("partition_data")
3179                    .transform(Transform::Identity)
3180                    .build()
3181                    .unwrap(),
3182            )
3183            .unwrap()
3184            .build();
3185
3186        let metadata = TableMetadataBuilder::new(
3187            initial_schema,
3188            partition_spec,
3189            SortOrder::unsorted_order(),
3190            TEST_LOCATION.to_string(),
3191            FormatVersion::V2,
3192            HashMap::new(),
3193        )
3194        .unwrap()
3195        .build()
3196        .unwrap()
3197        .metadata;
3198
3199        // Add a new schema that still contains the partition_data field
3200        let evolved_schema = Schema::builder()
3201            .with_fields(vec![
3202                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3203                NestedField::required(2, "partition_data", Type::Primitive(PrimitiveType::Int))
3204                    .into(),
3205                NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3206                    .into(),
3207            ])
3208            .build()
3209            .unwrap();
3210
3211        // This should succeed because partition_data exists in historical schemas
3212        let result = metadata
3213            .into_builder(Some("test_location".to_string()))
3214            .add_current_schema(evolved_schema);
3215
3216        assert!(result.is_ok());
3217    }
3218
3219    /// A v2 table of `data: string` partitioned by `bucket(data, 16)` as `bucket_data`.
3220    fn bucket_partitioned_metadata() -> TableMetadata {
3221        let schema = Schema::builder()
3222            .with_fields(vec![
3223                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3224            ])
3225            .build()
3226            .unwrap();
3227        let spec = UnboundPartitionSpec::builder()
3228            .with_spec_id(0)
3229            .add_partition_field(
3230                UnboundPartitionField::builder()
3231                    .source_ids(vec![1])
3232                    .name("bucket_data")
3233                    .transform(Transform::Bucket(16))
3234                    .build()
3235                    .unwrap(),
3236            )
3237            .unwrap()
3238            .build();
3239
3240        TableMetadataBuilder::new(
3241            schema,
3242            spec,
3243            SortOrder::unsorted_order(),
3244            TEST_LOCATION.to_string(),
3245            FormatVersion::V2,
3246            HashMap::new(),
3247        )
3248        .unwrap()
3249        .build()
3250        .unwrap()
3251        .metadata
3252    }
3253
3254    /// Replaces the default spec with an unpartitioned one. Spec 0 is retained but superseded.
3255    fn unpartition(metadata: TableMetadata, file: &str) -> TableMetadata {
3256        let metadata = metadata
3257            .into_builder(Some(file.to_string()))
3258            .add_default_partition_spec(UnboundPartitionSpec::builder().build())
3259            .unwrap()
3260            .build()
3261            .unwrap()
3262            .metadata;
3263        assert_ne!(metadata.default_partition_spec().spec_id(), 0);
3264        assert!(metadata.partition_specs.contains_key(&0));
3265        metadata
3266    }
3267
3268    #[test]
3269    fn test_schema_evolution_allows_a_superseded_partition_field_name_once_its_source_is_dropped() {
3270        // `data`, the source of spec 0's `bucket_data`, is gone, so spec 0 can only be historical
3271        // and its names are free. Iceberg Java skips its name check for exactly this case.
3272        let schema = Schema::builder()
3273            .with_fields(vec![
3274                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3275            ])
3276            .build()
3277            .unwrap();
3278
3279        let metadata = unpartition(
3280            bucket_partitioned_metadata(),
3281            "s3://bucket/t/metadata/m1.json",
3282        )
3283        .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3284        .add_current_schema(schema.clone())
3285        .unwrap()
3286        .build()
3287        .unwrap()
3288        .metadata;
3289
3290        assert_eq!(metadata.current_schema().as_struct(), schema.as_struct());
3291        assert!(metadata.current_schema().field_by_name("data").is_none());
3292        // Spec 0 is kept so the snapshots written under it stay interpretable.
3293        assert_eq!(
3294            metadata.partition_spec_by_id(0).unwrap().fields()[0].name,
3295            "bucket_data"
3296        );
3297    }
3298
3299    #[test]
3300    fn test_schema_evolution_rejects_a_partition_field_name_while_its_source_remains() {
3301        // `data` is still here, so `bucket_data` would be ambiguous. Iceberg Java rejects this
3302        // whether or not spec 0 is still the default, because it rebuilds every spec.
3303        let schema = Schema::builder()
3304            .with_fields(vec![
3305                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3306                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3307            ])
3308            .build()
3309            .unwrap();
3310
3311        let error = bucket_partitioned_metadata()
3312            .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3313            .add_current_schema(schema.clone())
3314            .unwrap_err();
3315        assert_eq!(error.kind(), ErrorKind::DataInvalid);
3316        assert!(
3317            error.message().contains("sourced from field id 1"),
3318            "{error}"
3319        );
3320
3321        let error = unpartition(
3322            bucket_partitioned_metadata(),
3323            "s3://bucket/t/metadata/m1.json",
3324        )
3325        .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3326        .add_current_schema(schema)
3327        .unwrap_err();
3328        assert_eq!(error.kind(), ErrorKind::DataInvalid);
3329        assert!(
3330            error.message().contains("sourced from field id 1"),
3331            "{error}"
3332        );
3333    }
3334
3335    #[test]
3336    fn test_schema_evolution_partition_name_check_is_order_independent() {
3337        // Dropping the source column and dropping the partitioning in one set of changes must
3338        // behave the same either way round.
3339        let schema = Schema::builder()
3340            .with_fields(vec![
3341                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3342            ])
3343            .build()
3344            .unwrap();
3345        let unpartitioned = UnboundPartitionSpec::builder().build();
3346
3347        let schema_first = bucket_partitioned_metadata()
3348            .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3349            .add_current_schema(schema.clone())
3350            .and_then(|b| b.add_default_partition_spec(unpartitioned.clone()))
3351            .and_then(|b| b.build());
3352        let spec_first = bucket_partitioned_metadata()
3353            .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3354            .add_default_partition_spec(unpartitioned)
3355            .and_then(|b| b.add_current_schema(schema))
3356            .and_then(|b| b.build());
3357
3358        assert!(schema_first.is_ok(), "{:?}", schema_first.err());
3359        assert!(spec_first.is_ok(), "{:?}", spec_first.err());
3360    }
3361
3362    #[test]
3363    fn test_set_current_schema_rejects_a_partition_field_name_from_a_spec_added_after_the_schema() {
3364        let schema = Schema::builder()
3365            .with_fields(vec![
3366                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3367            ])
3368            .build()
3369            .unwrap();
3370        let conflicting_schema = Schema::builder()
3371            .with_fields(vec![
3372                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3373                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3374            ])
3375            .build()
3376            .unwrap();
3377        let spec = UnboundPartitionSpec::builder()
3378            .add_partition_field(
3379                UnboundPartitionField::builder()
3380                    .source_ids(vec![1])
3381                    .name("bucket_data")
3382                    .transform(Transform::Bucket(16))
3383                    .build()
3384                    .unwrap(),
3385            )
3386            .unwrap()
3387            .build();
3388
3389        let error = TableMetadataBuilder::new(
3390            schema,
3391            UnboundPartitionSpec::builder().build(),
3392            SortOrder::unsorted_order(),
3393            TEST_LOCATION.to_string(),
3394            FormatVersion::V2,
3395            HashMap::new(),
3396        )
3397        .unwrap()
3398        .add_schema(conflicting_schema)
3399        .and_then(|b| b.add_partition_spec(spec))
3400        .and_then(|b| b.set_current_schema(TableMetadataBuilder::LAST_ADDED))
3401        .unwrap_err();
3402        assert_eq!(error.kind(), ErrorKind::DataInvalid);
3403        assert!(
3404            error.message().contains("sourced from field id 1"),
3405            "{error}"
3406        );
3407    }
3408
3409    #[test]
3410    fn test_schema_with_a_conflicting_partition_field_name_can_be_added_but_not_made_current() {
3411        let conflicting_schema = Schema::builder()
3412            .with_fields(vec![
3413                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3414                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3415            ])
3416            .build()
3417            .unwrap();
3418
3419        let metadata = bucket_partitioned_metadata()
3420            .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3421            .add_schema(conflicting_schema.clone())
3422            .unwrap()
3423            .build()
3424            .unwrap()
3425            .metadata;
3426        assert_eq!(metadata.current_schema_id(), 0);
3427        assert_eq!(
3428            metadata.schema_by_id(1).unwrap().as_struct(),
3429            conflicting_schema.as_struct()
3430        );
3431
3432        let error = metadata
3433            .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3434            .set_current_schema(1)
3435            .unwrap_err();
3436        assert_eq!(error.kind(), ErrorKind::DataInvalid);
3437        assert!(
3438            error.message().contains("sourced from field id 1"),
3439            "{error}"
3440        );
3441    }
3442
3443    #[test]
3444    fn test_schema_evolution_prevents_new_field_conflicting_with_partition_field() {
3445        // Create initial schema WITHOUT the conflicting field
3446        let initial_schema = Schema::builder()
3447            .with_fields(vec![
3448                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3449            ])
3450            .build()
3451            .unwrap();
3452
3453        let partition_spec = UnboundPartitionSpec::builder()
3454            .with_spec_id(0)
3455            .add_partition_field(
3456                UnboundPartitionField::builder()
3457                    .source_ids(vec![1])
3458                    .name("bucket_data")
3459                    .transform(Transform::Bucket(16))
3460                    .build()
3461                    .unwrap(),
3462            )
3463            .unwrap()
3464            .build();
3465
3466        let metadata = TableMetadataBuilder::new(
3467            initial_schema,
3468            partition_spec,
3469            SortOrder::unsorted_order(),
3470            TEST_LOCATION.to_string(),
3471            FormatVersion::V2,
3472            HashMap::new(),
3473        )
3474        .unwrap()
3475        .build()
3476        .unwrap()
3477        .metadata;
3478
3479        // Try to add a schema with a field that conflicts with partition field name
3480        let conflicting_schema = Schema::builder()
3481            .with_fields(vec![
3482                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3483                // This field name conflicts with the partition field "bucket_data"
3484                NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3485            ])
3486            .build()
3487            .unwrap();
3488
3489        let builder = metadata.into_builder(Some("test_location".to_string()));
3490        let result = builder.add_current_schema(conflicting_schema);
3491
3492        // This should fail because "bucket_data" takes the name of a partition field whose
3493        // source column `data` is still in the schema
3494        assert!(result.is_err());
3495        let error = result.unwrap_err();
3496        assert!(error.message().contains("Cannot set current schema"));
3497        assert!(error.message().contains("sourced from field id 1"));
3498    }
3499
3500    #[test]
3501    fn test_partition_spec_evolution_allows_non_conflicting_names() {
3502        let initial_schema = Schema::builder()
3503            .with_fields(vec![
3504                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3505                NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
3506                    .into(),
3507            ])
3508            .build()
3509            .unwrap();
3510
3511        let partition_spec = UnboundPartitionSpec::builder()
3512            .with_spec_id(0)
3513            .add_partition_field(
3514                UnboundPartitionField::builder()
3515                    .source_ids(vec![1])
3516                    .name("data_bucket")
3517                    .transform(Transform::Bucket(16))
3518                    .build()
3519                    .unwrap(),
3520            )
3521            .unwrap()
3522            .build();
3523
3524        let metadata = TableMetadataBuilder::new(
3525            initial_schema,
3526            partition_spec,
3527            SortOrder::unsorted_order(),
3528            TEST_LOCATION.to_string(),
3529            FormatVersion::V2,
3530            HashMap::new(),
3531        )
3532        .unwrap()
3533        .build()
3534        .unwrap()
3535        .metadata;
3536
3537        let builder = metadata.into_builder(Some(
3538            "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3539        ));
3540
3541        // Try to add a partition spec with a field name that does NOT conflict with existing schema fields
3542        let non_conflicting_partition_spec = UnboundPartitionSpec::builder()
3543            .with_spec_id(1)
3544            .add_partition_field(
3545                UnboundPartitionField::builder()
3546                    .source_ids(vec![2])
3547                    .name("new_partition_field")
3548                    .transform(Transform::Bucket(8))
3549                    .build()
3550                    .unwrap(),
3551            )
3552            .unwrap()
3553            .build();
3554
3555        let result = builder.add_partition_spec(non_conflicting_partition_spec);
3556
3557        assert!(result.is_ok());
3558    }
3559
3560    #[test]
3561    fn test_row_lineage_addition() {
3562        let new_rows = 30;
3563        let base = builder_without_changes(FormatVersion::V3)
3564            .build()
3565            .unwrap()
3566            .metadata;
3567        let add_rows = Snapshot::builder()
3568            .with_snapshot_id(0)
3569            .with_timestamp_ms(base.last_updated_ms + 1)
3570            .with_sequence_number(0)
3571            .with_schema_id(0)
3572            .with_manifest_list("foo")
3573            .with_parent_snapshot_id(None)
3574            .with_summary(Summary {
3575                operation: Operation::Append,
3576                additional_properties: HashMap::new(),
3577            })
3578            .with_row_range(base.next_row_id(), new_rows)
3579            .build();
3580
3581        let first_addition = base
3582            .into_builder(None)
3583            .add_snapshot(add_rows.clone())
3584            .unwrap()
3585            .build()
3586            .unwrap()
3587            .metadata;
3588
3589        assert_eq!(first_addition.next_row_id(), new_rows);
3590
3591        let add_more_rows = Snapshot::builder()
3592            .with_snapshot_id(1)
3593            .with_timestamp_ms(first_addition.last_updated_ms + 1)
3594            .with_sequence_number(1)
3595            .with_schema_id(0)
3596            .with_manifest_list("foo")
3597            .with_parent_snapshot_id(Some(0))
3598            .with_summary(Summary {
3599                operation: Operation::Append,
3600                additional_properties: HashMap::new(),
3601            })
3602            .with_row_range(first_addition.next_row_id(), new_rows)
3603            .build();
3604
3605        let second_addition = first_addition
3606            .into_builder(None)
3607            .add_snapshot(add_more_rows)
3608            .unwrap()
3609            .build()
3610            .unwrap()
3611            .metadata;
3612        assert_eq!(second_addition.next_row_id(), new_rows * 2);
3613    }
3614
3615    #[test]
3616    fn test_row_lineage_invalid_snapshot() {
3617        let new_rows = 30;
3618        let base = builder_without_changes(FormatVersion::V3)
3619            .build()
3620            .unwrap()
3621            .metadata;
3622
3623        // add rows to check TableMetadata validation; Snapshot rejects negative next-row-id
3624        let add_rows = Snapshot::builder()
3625            .with_snapshot_id(0)
3626            .with_timestamp_ms(base.last_updated_ms + 1)
3627            .with_sequence_number(0)
3628            .with_schema_id(0)
3629            .with_manifest_list("foo")
3630            .with_parent_snapshot_id(None)
3631            .with_summary(Summary {
3632                operation: Operation::Append,
3633                additional_properties: HashMap::new(),
3634            })
3635            .with_row_range(base.next_row_id(), new_rows)
3636            .build();
3637
3638        let added = base
3639            .into_builder(None)
3640            .add_snapshot(add_rows)
3641            .unwrap()
3642            .build()
3643            .unwrap()
3644            .metadata;
3645
3646        let invalid_new_rows = Snapshot::builder()
3647            .with_snapshot_id(1)
3648            .with_timestamp_ms(added.last_updated_ms + 1)
3649            .with_sequence_number(1)
3650            .with_schema_id(0)
3651            .with_manifest_list("foo")
3652            .with_parent_snapshot_id(Some(0))
3653            .with_summary(Summary {
3654                operation: Operation::Append,
3655                additional_properties: HashMap::new(),
3656            })
3657            // first_row_id is behind table next_row_id
3658            .with_row_range(added.next_row_id() - 1, 10)
3659            .build();
3660
3661        let err = added
3662            .into_builder(None)
3663            .add_snapshot(invalid_new_rows)
3664            .unwrap_err();
3665        assert!(
3666            err.to_string().contains(
3667                "Cannot add a snapshot, first-row-id is behind table next-row-id: 29 < 30"
3668            )
3669        );
3670    }
3671
3672    #[test]
3673    fn test_row_lineage_append_branch() {
3674        // Appends to a branch should still change last-row-id even if not on main, these changes
3675        // should also affect commits to main
3676
3677        let branch = "some_branch";
3678
3679        // Start with V3 metadata to support row lineage
3680        let base = builder_without_changes(FormatVersion::V3)
3681            .build()
3682            .unwrap()
3683            .metadata;
3684
3685        // Initial next_row_id should be 0
3686        assert_eq!(base.next_row_id(), 0);
3687
3688        // Write to Branch - append 30 rows
3689        let branch_snapshot_1 = Snapshot::builder()
3690            .with_snapshot_id(1)
3691            .with_timestamp_ms(base.last_updated_ms + 1)
3692            .with_sequence_number(0)
3693            .with_schema_id(0)
3694            .with_manifest_list("foo")
3695            .with_parent_snapshot_id(None)
3696            .with_summary(Summary {
3697                operation: Operation::Append,
3698                additional_properties: HashMap::new(),
3699            })
3700            .with_row_range(base.next_row_id(), 30)
3701            .build();
3702
3703        let table_after_branch_1 = base
3704            .into_builder(None)
3705            .set_branch_snapshot(branch_snapshot_1.clone(), branch)
3706            .unwrap()
3707            .build()
3708            .unwrap()
3709            .metadata;
3710
3711        // Current snapshot should be null (no main branch snapshot yet)
3712        assert!(table_after_branch_1.current_snapshot().is_none());
3713
3714        // Branch snapshot should have first_row_id = 0
3715        let branch_ref = table_after_branch_1.refs.get(branch).unwrap();
3716        let branch_snap_1 = table_after_branch_1
3717            .snapshots
3718            .get(&branch_ref.snapshot_id)
3719            .unwrap();
3720        assert_eq!(branch_snap_1.first_row_id(), Some(0));
3721
3722        // Next row id should be 30
3723        assert_eq!(table_after_branch_1.next_row_id(), 30);
3724
3725        // Write to Main - append 28 rows
3726        let main_snapshot = Snapshot::builder()
3727            .with_snapshot_id(2)
3728            .with_timestamp_ms(table_after_branch_1.last_updated_ms + 1)
3729            .with_sequence_number(1)
3730            .with_schema_id(0)
3731            .with_manifest_list("bar")
3732            .with_parent_snapshot_id(None)
3733            .with_summary(Summary {
3734                operation: Operation::Append,
3735                additional_properties: HashMap::new(),
3736            })
3737            .with_row_range(table_after_branch_1.next_row_id(), 28)
3738            .build();
3739
3740        let table_after_main = table_after_branch_1
3741            .into_builder(None)
3742            .add_snapshot(main_snapshot.clone())
3743            .unwrap()
3744            .set_ref(MAIN_BRANCH, SnapshotReference {
3745                snapshot_id: main_snapshot.snapshot_id(),
3746                retention: SnapshotRetention::Branch {
3747                    min_snapshots_to_keep: None,
3748                    max_snapshot_age_ms: None,
3749                    max_ref_age_ms: None,
3750                },
3751            })
3752            .unwrap()
3753            .build()
3754            .unwrap()
3755            .metadata;
3756
3757        // Main snapshot should have first_row_id = 30
3758        let current_snapshot = table_after_main.current_snapshot().unwrap();
3759        assert_eq!(current_snapshot.first_row_id(), Some(30));
3760
3761        // Next row id should be 58 (30 + 28)
3762        assert_eq!(table_after_main.next_row_id(), 58);
3763
3764        // Write again to branch - append 21 rows
3765        let branch_snapshot_2 = Snapshot::builder()
3766            .with_snapshot_id(3)
3767            .with_timestamp_ms(table_after_main.last_updated_ms + 1)
3768            .with_sequence_number(2)
3769            .with_schema_id(0)
3770            .with_manifest_list("baz")
3771            .with_parent_snapshot_id(Some(branch_snapshot_1.snapshot_id()))
3772            .with_summary(Summary {
3773                operation: Operation::Append,
3774                additional_properties: HashMap::new(),
3775            })
3776            .with_row_range(table_after_main.next_row_id(), 21)
3777            .build();
3778
3779        let table_after_branch_2 = table_after_main
3780            .into_builder(None)
3781            .set_branch_snapshot(branch_snapshot_2.clone(), branch)
3782            .unwrap()
3783            .build()
3784            .unwrap()
3785            .metadata;
3786
3787        // Branch snapshot should have first_row_id = 58 (30 + 28)
3788        let branch_ref_2 = table_after_branch_2.refs.get(branch).unwrap();
3789        let branch_snap_2 = table_after_branch_2
3790            .snapshots
3791            .get(&branch_ref_2.snapshot_id)
3792            .unwrap();
3793        assert_eq!(branch_snap_2.first_row_id(), Some(58));
3794
3795        // Next row id should be 79 (30 + 28 + 21)
3796        assert_eq!(table_after_branch_2.next_row_id(), 79);
3797    }
3798
3799    #[test]
3800    fn test_encryption_keys() {
3801        let builder = builder_without_changes(FormatVersion::V2);
3802
3803        // Create test encryption keys
3804        let encryption_key_1 = EncryptedKey::builder()
3805            .key_id("key-1")
3806            .encrypted_key_metadata(vec![1, 2, 3, 4])
3807            .encrypted_by_id("encryption-service-1")
3808            .properties(HashMap::from_iter(vec![(
3809                "algorithm".to_string(),
3810                "AES-256".to_string(),
3811            )]))
3812            .build();
3813
3814        let encryption_key_2 = EncryptedKey::builder()
3815            .key_id("key-2")
3816            .encrypted_key_metadata(vec![5, 6, 7, 8])
3817            .encrypted_by_id("encryption-service-2")
3818            .properties(HashMap::new())
3819            .build();
3820
3821        // Add first encryption key
3822        let build_result = builder
3823            .add_encryption_key(encryption_key_1.clone())
3824            .build()
3825            .unwrap();
3826
3827        assert_eq!(build_result.changes.len(), 1);
3828        assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3829        assert_eq!(
3830            build_result.metadata.encryption_key("key-1"),
3831            Some(&encryption_key_1)
3832        );
3833        assert_eq!(build_result.changes[0], TableUpdate::AddEncryptionKey {
3834            encryption_key: encryption_key_1.clone()
3835        });
3836
3837        // Add second encryption key
3838        let build_result = build_result
3839            .metadata
3840            .into_builder(Some(
3841                "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3842            ))
3843            .add_encryption_key(encryption_key_2.clone())
3844            .build()
3845            .unwrap();
3846
3847        assert_eq!(build_result.changes.len(), 1);
3848        assert_eq!(build_result.metadata.encryption_keys.len(), 2);
3849        assert_eq!(
3850            build_result.metadata.encryption_key("key-1"),
3851            Some(&encryption_key_1)
3852        );
3853        assert_eq!(
3854            build_result.metadata.encryption_key("key-2"),
3855            Some(&encryption_key_2)
3856        );
3857        assert_eq!(build_result.changes[0], TableUpdate::AddEncryptionKey {
3858            encryption_key: encryption_key_2.clone()
3859        });
3860
3861        // Try to add duplicate key - should not create a change
3862        let build_result = build_result
3863            .metadata
3864            .into_builder(Some(
3865                "s3://bucket/test/location/metadata/metadata2.json".to_string(),
3866            ))
3867            .add_encryption_key(encryption_key_1.clone())
3868            .build()
3869            .unwrap();
3870
3871        assert_eq!(build_result.changes.len(), 0);
3872        assert_eq!(build_result.metadata.encryption_keys.len(), 2);
3873
3874        // Remove first encryption key
3875        let build_result = build_result
3876            .metadata
3877            .into_builder(Some(
3878                "s3://bucket/test/location/metadata/metadata3.json".to_string(),
3879            ))
3880            .remove_encryption_key("key-1")
3881            .build()
3882            .unwrap();
3883
3884        assert_eq!(build_result.changes.len(), 1);
3885        assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3886        assert_eq!(build_result.metadata.encryption_key("key-1"), None);
3887        assert_eq!(
3888            build_result.metadata.encryption_key("key-2"),
3889            Some(&encryption_key_2)
3890        );
3891        assert_eq!(build_result.changes[0], TableUpdate::RemoveEncryptionKey {
3892            key_id: "key-1".to_string()
3893        });
3894
3895        // Try to remove non-existent key - should not create a change
3896        let build_result = build_result
3897            .metadata
3898            .into_builder(Some(
3899                "s3://bucket/test/location/metadata/metadata4.json".to_string(),
3900            ))
3901            .remove_encryption_key("non-existent-key")
3902            .build()
3903            .unwrap();
3904
3905        assert_eq!(build_result.changes.len(), 0);
3906        assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3907
3908        // Test encryption_keys_iter()
3909        let keys = build_result
3910            .metadata
3911            .encryption_keys_iter()
3912            .collect::<Vec<_>>();
3913        assert_eq!(keys.len(), 1);
3914        assert_eq!(keys[0], &encryption_key_2);
3915
3916        // Remove last encryption key
3917        let build_result = build_result
3918            .metadata
3919            .into_builder(Some(
3920                "s3://bucket/test/location/metadata/metadata5.json".to_string(),
3921            ))
3922            .remove_encryption_key("key-2")
3923            .build()
3924            .unwrap();
3925
3926        assert_eq!(build_result.changes.len(), 1);
3927        assert_eq!(build_result.metadata.encryption_keys.len(), 0);
3928        assert_eq!(build_result.metadata.encryption_key("key-2"), None);
3929        assert_eq!(build_result.changes[0], TableUpdate::RemoveEncryptionKey {
3930            key_id: "key-2".to_string()
3931        });
3932
3933        // Verify empty encryption_keys_iter()
3934        let keys = build_result.metadata.encryption_keys_iter();
3935        assert_eq!(keys.len(), 0);
3936    }
3937
3938    #[test]
3939    fn test_partition_field_id_reuse_across_specs() {
3940        let schema = Schema::builder()
3941            .with_fields(vec![
3942                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
3943                NestedField::required(2, "data", Type::Primitive(PrimitiveType::String)).into(),
3944                NestedField::required(3, "timestamp", Type::Primitive(PrimitiveType::Timestamp))
3945                    .into(),
3946            ])
3947            .build()
3948            .unwrap();
3949
3950        // Create initial table with spec 0: identity(id) -> field_id = 1000
3951        let initial_spec = UnboundPartitionSpec::builder()
3952            .add_partition_field(
3953                UnboundPartitionField::builder()
3954                    .source_ids(vec![1])
3955                    .name("id")
3956                    .transform(Transform::Identity)
3957                    .build()
3958                    .unwrap(),
3959            )
3960            .unwrap()
3961            .build();
3962
3963        let mut metadata = TableMetadataBuilder::new(
3964            schema,
3965            initial_spec,
3966            SortOrder::unsorted_order(),
3967            "s3://bucket/table".to_string(),
3968            FormatVersion::V2,
3969            HashMap::new(),
3970        )
3971        .unwrap()
3972        .build()
3973        .unwrap()
3974        .metadata;
3975
3976        // Add spec 1: bucket(data) -> field_id = 1001
3977        let spec1 = UnboundPartitionSpec::builder()
3978            .add_partition_field(
3979                UnboundPartitionField::builder()
3980                    .source_ids(vec![2])
3981                    .name("data_bucket")
3982                    .transform(Transform::Bucket(10))
3983                    .build()
3984                    .unwrap(),
3985            )
3986            .unwrap()
3987            .build();
3988        let builder = metadata.into_builder(Some("s3://bucket/table/metadata/v1.json".to_string()));
3989        let result = builder.add_partition_spec(spec1).unwrap().build().unwrap();
3990        metadata = result.metadata;
3991
3992        // Add spec 2: identity(id) + bucket(data) + year(timestamp)
3993        // Should reuse field_id 1000 for identity(id) and 1001 for bucket(data)
3994        let spec2 = UnboundPartitionSpec::builder()
3995            .add_partition_field(
3996                UnboundPartitionField::builder()
3997                    .source_ids(vec![1])
3998                    .name("id")
3999                    .transform(Transform::Identity)
4000                    .build()
4001                    .unwrap(),
4002            ) // Should reuse 1000
4003            .unwrap()
4004            .add_partition_field(
4005                UnboundPartitionField::builder()
4006                    .source_ids(vec![2])
4007                    .name("data_bucket")
4008                    .transform(Transform::Bucket(10))
4009                    .build()
4010                    .unwrap(),
4011            ) // Should reuse 1001
4012            .unwrap()
4013            .add_partition_field(
4014                UnboundPartitionField::builder()
4015                    .source_ids(vec![3])
4016                    .name("year")
4017                    .transform(Transform::Year)
4018                    .build()
4019                    .unwrap(),
4020            ) // Should get new 1002
4021            .unwrap()
4022            .build();
4023        let builder = metadata.into_builder(Some("s3://bucket/table/metadata/v2.json".to_string()));
4024        let result = builder.add_partition_spec(spec2).unwrap().build().unwrap();
4025
4026        // Verify field ID reuse: spec 2 should reuse IDs from specs 0 and 1, assign new ID for new field
4027        let spec2 = result.metadata.partition_spec_by_id(2).unwrap();
4028        let field_ids: Vec<i32> = spec2.fields().iter().map(|f| f.field_id).collect();
4029        assert_eq!(field_ids, vec![1000, 1001, 1002]); // Reused 1000, 1001; new 1002
4030    }
4031
4032    #[test]
4033    fn test_highest_equivalent_field_ids_is_independent_of_field_order() {
4034        let field = |source_id, field_id, transform| PartitionField {
4035            source_id,
4036            field_id,
4037            name: format!("f{field_id}"),
4038            transform,
4039        };
4040        let fields = [
4041            field(1, 1000, Transform::Identity),
4042            field(2, 1003, Transform::Identity),
4043            field(1, 1006, Transform::Identity),
4044            field(1, 1005, Transform::Bucket(16)),
4045            field(2, 1001, Transform::Identity),
4046            field(1, 1002, Transform::Identity),
4047            field(1, 1004, Transform::Bucket(16)),
4048        ];
4049        let expected = HashMap::from([
4050            ((vec![1], Transform::Identity), 1006),
4051            ((vec![2], Transform::Identity), 1003),
4052            ((vec![1], Transform::Bucket(16)), 1005),
4053        ]);
4054
4055        assert_eq!(
4056            TableMetadataBuilder::highest_equivalent_field_ids(fields.iter()),
4057            expected
4058        );
4059        assert_eq!(
4060            TableMetadataBuilder::highest_equivalent_field_ids(fields.iter().rev()),
4061            expected
4062        );
4063    }
4064
4065    #[test]
4066    fn test_partition_field_id_reuse_picks_the_highest_of_several_candidates() {
4067        // These specs give identity(x) four different field ids, which a writer that does not
4068        // reuse ids can leave behind. `partition_specs` is a HashMap, so the id reused for a new
4069        // spec must not depend on which of them is visited first.
4070        let metadata: TableMetadata = serde_json::from_value(serde_json::json!({
4071            "format-version": 2,
4072            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
4073            "location": "s3://bucket/test/location",
4074            "last-sequence-number": 0,
4075            "last-updated-ms": 1602638573590i64,
4076            "last-column-id": 2,
4077            "current-schema-id": 0,
4078            "schemas": [{
4079                "type": "struct",
4080                "schema-id": 0,
4081                "fields": [
4082                    {"id": 1, "name": "x", "required": true, "type": "long"},
4083                    {"id": 2, "name": "y", "required": true, "type": "long"},
4084                ],
4085            }],
4086            "default-spec-id": 3,
4087            "last-partition-id": 1006,
4088            "partition-specs": [
4089                {"spec-id": 0, "fields": [
4090                    {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000},
4091                ]},
4092                {"spec-id": 1, "fields": [
4093                    {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1002},
4094                ]},
4095                {"spec-id": 2, "fields": [
4096                    {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1004},
4097                ]},
4098                {"spec-id": 3, "fields": [
4099                    {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1006},
4100                ]},
4101            ],
4102            "default-sort-order-id": 0,
4103            "sort-orders": [{"order-id": 0, "fields": []}],
4104        }))
4105        .unwrap();
4106
4107        // Two fields, so this is not compatible with any existing spec and is added as a new
4108        // one rather than resolving to an existing spec id.
4109        let new_spec = UnboundPartitionSpec::builder()
4110            .add_partition_field(
4111                UnboundPartitionField::builder()
4112                    .source_ids(vec![1])
4113                    .name("x")
4114                    .transform(Transform::Identity)
4115                    .build()
4116                    .unwrap(),
4117            )
4118            .unwrap()
4119            .add_partition_field(
4120                UnboundPartitionField::builder()
4121                    .source_ids(vec![2])
4122                    .name("y")
4123                    .transform(Transform::Identity)
4124                    .build()
4125                    .unwrap(),
4126            )
4127            .unwrap()
4128            .build();
4129
4130        let result = metadata
4131            .into_builder(Some(
4132                "s3://bucket/test/location/metadata/v1.json".to_string(),
4133            ))
4134            .add_partition_spec(new_spec)
4135            .unwrap()
4136            .build()
4137            .unwrap();
4138
4139        let spec = result.metadata.partition_spec_by_id(4).unwrap();
4140        let fields: Vec<(i32, i32)> = spec
4141            .fields()
4142            .iter()
4143            .map(|f| (f.source_id, f.field_id))
4144            .collect();
4145        assert_eq!(fields, vec![(1, 1006), (2, 1007)]);
4146        assert_eq!(result.metadata.last_partition_id, 1007);
4147    }
4148}