Skip to main content

iceberg/spec/
table_metadata.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
18//! Defines the [table metadata](https://iceberg.apache.org/spec/#table-metadata).
19//! The main struct here is [TableMetadataV2] which defines the data for a table.
20
21use std::cmp::Ordering;
22use std::collections::HashMap;
23use std::fmt::{Display, Formatter};
24use std::hash::Hash;
25use std::sync::Arc;
26
27use _serde::TableMetadataEnum;
28use chrono::{DateTime, Utc};
29use serde::{Deserialize, Serialize};
30use serde_repr::{Deserialize_repr, Serialize_repr};
31use uuid::Uuid;
32
33use super::snapshot::SnapshotReference;
34pub use super::table_metadata_builder::{TableMetadataBuildResult, TableMetadataBuilder};
35use super::{
36    DEFAULT_PARTITION_SPEC_ID, PartitionSpecRef, PartitionStatisticsFile, SchemaId, SchemaRef,
37    SnapshotRef, SnapshotRetention, SortOrder, SortOrderRef, StatisticsFile, StructType,
38    TableProperties, parse_metadata_file_compression,
39};
40use crate::catalog::{METADATA_FOLDER_NAME, MetadataLocation};
41use crate::compression::CompressionCodec;
42use crate::error::{Result, timestamp_ms_to_utc};
43use crate::io::FileIO;
44use crate::spec::EncryptedKey;
45use crate::{Error, ErrorKind};
46
47static MAIN_BRANCH: &str = "main";
48pub(crate) static ONE_MINUTE_MS: i64 = 60_000;
49
50/// Sentinel value used by the Java implementation and older metadata files
51/// to represent a missing/empty current snapshot ID. During deserialization,
52/// this value is normalized to `None`.
53pub(crate) static EMPTY_SNAPSHOT_ID: i64 = -1;
54pub(crate) static INITIAL_SEQUENCE_NUMBER: i64 = 0;
55
56/// Initial row id for row lineage for new v3 tables and older tables upgrading to v3.
57pub const INITIAL_ROW_ID: u64 = 0;
58/// Minimum format version that supports row lineage (v3).
59pub const MIN_FORMAT_VERSION_ROW_LINEAGE: FormatVersion = FormatVersion::V3;
60/// Reference to [`TableMetadata`].
61pub type TableMetadataRef = Arc<TableMetadata>;
62
63#[derive(Debug, PartialEq, Deserialize, Eq, Clone)]
64#[serde(try_from = "TableMetadataEnum")]
65/// Fields for the version 2 of the table metadata.
66///
67/// We assume that this data structure is always valid, so we will panic when invalid error happens.
68/// We check the validity of this data structure when constructing.
69pub struct TableMetadata {
70    /// Integer Version for the format.
71    pub(crate) format_version: FormatVersion,
72    /// A UUID that identifies the table
73    pub(crate) table_uuid: Uuid,
74    /// Location tables base location
75    pub(crate) location: String,
76    /// The tables highest sequence number
77    pub(crate) last_sequence_number: i64,
78    /// Timestamp in milliseconds from the unix epoch when the table was last updated.
79    pub(crate) last_updated_ms: i64,
80    /// An integer; the highest assigned column ID for the table.
81    pub(crate) last_column_id: i32,
82    /// A list of schemas, stored as objects with schema-id.
83    pub(crate) schemas: HashMap<i32, SchemaRef>,
84    /// ID of the table’s current schema.
85    pub(crate) current_schema_id: i32,
86    /// A list of partition specs, stored as full partition spec objects.
87    pub(crate) partition_specs: HashMap<i32, PartitionSpecRef>,
88    /// ID of the “current” spec that writers should use by default.
89    pub(crate) default_spec: PartitionSpecRef,
90    /// Partition type of the default partition spec.
91    pub(crate) default_partition_type: StructType,
92    /// An integer; the highest assigned partition field ID across all partition specs for the table.
93    pub(crate) last_partition_id: i32,
94    ///A string to string map of table properties. This is used to control settings that
95    /// affect reading and writing and is not intended to be used for arbitrary metadata.
96    /// For example, commit.retry.num-retries is used to control the number of commit retries.
97    pub(crate) properties: HashMap<String, String>,
98    /// long ID of the current table snapshot; must be the same as the current
99    /// ID of the main branch in refs.
100    pub(crate) current_snapshot_id: Option<i64>,
101    ///A list of valid snapshots. Valid snapshots are snapshots for which all
102    /// data files exist in the file system. A data file must not be deleted
103    /// from the file system until the last snapshot in which it was listed is
104    /// garbage collected.
105    pub(crate) snapshots: HashMap<i64, SnapshotRef>,
106    /// A list (optional) of timestamp and snapshot ID pairs that encodes changes
107    /// to the current snapshot for the table. Each time the current-snapshot-id
108    /// is changed, a new entry should be added with the last-updated-ms
109    /// and the new current-snapshot-id. When snapshots are expired from
110    /// the list of valid snapshots, all entries before a snapshot that has
111    /// expired should be removed.
112    pub(crate) snapshot_log: Vec<SnapshotLog>,
113
114    /// A list (optional) of timestamp and metadata file location pairs
115    /// that encodes changes to the previous metadata files for the table.
116    /// Each time a new metadata file is created, a new entry of the
117    /// previous metadata file location should be added to the list.
118    /// Tables can be configured to remove the oldest metadata log entries and
119    /// keep a fixed-size log of the most recent entries after a commit.
120    pub(crate) metadata_log: Vec<MetadataLog>,
121
122    /// A list of sort orders, stored as full sort order objects.
123    pub(crate) sort_orders: HashMap<i64, SortOrderRef>,
124    /// Default sort order id of the table. Note that this could be used by
125    /// writers, but is not used when reading because reads use the specs
126    /// stored in manifest files.
127    pub(crate) default_sort_order_id: i64,
128    /// A map of snapshot references. The map keys are the unique snapshot reference
129    /// names in the table, and the map values are snapshot reference objects.
130    /// There is always a main branch reference pointing to the current-snapshot-id
131    /// even if the refs map is null.
132    pub(crate) refs: HashMap<String, SnapshotReference>,
133    /// Mapping of snapshot ids to statistics files.
134    pub(crate) statistics: HashMap<i64, StatisticsFile>,
135    /// Mapping of snapshot ids to partition statistics files.
136    pub(crate) partition_statistics: HashMap<i64, PartitionStatisticsFile>,
137    /// Encryption Keys - map of key id to the actual key
138    pub(crate) encryption_keys: HashMap<String, EncryptedKey>,
139    /// Next row id to be assigned for Row Lineage (v3)
140    pub(crate) next_row_id: u64,
141}
142
143impl TableMetadata {
144    /// Convert this Table Metadata into a builder for modification.
145    ///
146    /// `current_file_location` is the location where the current version
147    /// of the metadata file is stored. This is used to update the metadata log.
148    /// If `current_file_location` is `None`, the metadata log will not be updated.
149    /// This should only be used to stage-create tables.
150    #[must_use]
151    pub fn into_builder(self, current_file_location: Option<String>) -> TableMetadataBuilder {
152        TableMetadataBuilder::new_from_metadata(self, current_file_location)
153    }
154
155    /// Check if a partition field name exists in any partition spec.
156    #[inline]
157    pub(crate) fn partition_name_exists(&self, name: &str) -> bool {
158        self.partition_specs
159            .values()
160            .any(|spec| spec.fields().iter().any(|pf| pf.name == name))
161    }
162
163    /// Check if a field name exists in any schema.
164    #[inline]
165    pub(crate) fn name_exists_in_any_schema(&self, name: &str) -> bool {
166        self.schemas
167            .values()
168            .any(|schema| schema.field_by_name(name).is_some())
169    }
170
171    /// Returns format version of this metadata.
172    #[inline]
173    pub fn format_version(&self) -> FormatVersion {
174        self.format_version
175    }
176
177    /// Returns uuid of current table.
178    #[inline]
179    pub fn uuid(&self) -> Uuid {
180        self.table_uuid
181    }
182
183    /// Returns table location.
184    #[inline]
185    pub fn location(&self) -> &str {
186        self.location.as_str()
187    }
188
189    /// Returns last sequence number.
190    #[inline]
191    pub fn last_sequence_number(&self) -> i64 {
192        self.last_sequence_number
193    }
194
195    /// Returns the next sequence number for the table.
196    ///
197    /// For format version 1, it always returns the initial sequence number.
198    /// For other versions, it returns the last sequence number incremented by 1.
199    #[inline]
200    pub fn next_sequence_number(&self) -> i64 {
201        match self.format_version {
202            FormatVersion::V1 => INITIAL_SEQUENCE_NUMBER,
203            _ => self.last_sequence_number + 1,
204        }
205    }
206
207    /// Returns the last column id.
208    #[inline]
209    pub fn last_column_id(&self) -> i32 {
210        self.last_column_id
211    }
212
213    /// Returns the last partition_id
214    #[inline]
215    pub fn last_partition_id(&self) -> i32 {
216        self.last_partition_id
217    }
218
219    /// Returns last updated time.
220    #[inline]
221    pub fn last_updated_timestamp(&self) -> Result<DateTime<Utc>> {
222        timestamp_ms_to_utc(self.last_updated_ms)
223    }
224
225    /// Returns last updated time in milliseconds.
226    #[inline]
227    pub fn last_updated_ms(&self) -> i64 {
228        self.last_updated_ms
229    }
230
231    /// Returns schemas
232    #[inline]
233    pub fn schemas_iter(&self) -> impl ExactSizeIterator<Item = &SchemaRef> {
234        self.schemas.values()
235    }
236
237    /// Lookup schema by id.
238    #[inline]
239    pub fn schema_by_id(&self, schema_id: SchemaId) -> Option<&SchemaRef> {
240        self.schemas.get(&schema_id)
241    }
242
243    /// Get current schema
244    #[inline]
245    pub fn current_schema(&self) -> &SchemaRef {
246        self.schema_by_id(self.current_schema_id)
247            .expect("Current schema id set, but not found in table metadata")
248    }
249
250    /// Get the id of the current schema
251    #[inline]
252    pub fn current_schema_id(&self) -> SchemaId {
253        self.current_schema_id
254    }
255
256    /// Returns all partition specs.
257    #[inline]
258    pub fn partition_specs_iter(&self) -> impl ExactSizeIterator<Item = &PartitionSpecRef> {
259        self.partition_specs.values()
260    }
261
262    /// Lookup partition spec by id.
263    #[inline]
264    pub fn partition_spec_by_id(&self, spec_id: i32) -> Option<&PartitionSpecRef> {
265        self.partition_specs.get(&spec_id)
266    }
267
268    /// Get default partition spec
269    #[inline]
270    pub fn default_partition_spec(&self) -> &PartitionSpecRef {
271        &self.default_spec
272    }
273
274    /// Return the partition type of the default partition spec.
275    #[inline]
276    pub fn default_partition_type(&self) -> &StructType {
277        &self.default_partition_type
278    }
279
280    #[inline]
281    /// Returns spec id of the "current" partition spec.
282    pub fn default_partition_spec_id(&self) -> i32 {
283        self.default_spec.spec_id()
284    }
285
286    /// Returns all snapshots
287    #[inline]
288    pub fn snapshots(&self) -> impl ExactSizeIterator<Item = &SnapshotRef> {
289        self.snapshots.values()
290    }
291
292    /// Lookup snapshot by id.
293    #[inline]
294    pub fn snapshot_by_id(&self, snapshot_id: i64) -> Option<&SnapshotRef> {
295        self.snapshots.get(&snapshot_id)
296    }
297
298    /// Returns snapshot history.
299    #[inline]
300    pub fn history(&self) -> &[SnapshotLog] {
301        &self.snapshot_log
302    }
303
304    /// Returns the metadata log.
305    #[inline]
306    pub fn metadata_log(&self) -> &[MetadataLog] {
307        &self.metadata_log
308    }
309
310    /// Get current snapshot
311    #[inline]
312    pub fn current_snapshot(&self) -> Option<&SnapshotRef> {
313        self.current_snapshot_id.map(|s| {
314            self.snapshot_by_id(s)
315                .expect("Current snapshot id has been set, but doesn't exist in metadata")
316        })
317    }
318
319    /// Get the current snapshot id
320    #[inline]
321    pub fn current_snapshot_id(&self) -> Option<i64> {
322        self.current_snapshot_id
323    }
324
325    /// Get the snapshot for a reference
326    /// Returns an option if the `ref_name` is not found
327    #[inline]
328    pub fn snapshot_for_ref(&self, ref_name: &str) -> Option<&SnapshotRef> {
329        self.refs.get(ref_name).map(|r| {
330            self.snapshot_by_id(r.snapshot_id)
331                .unwrap_or_else(|| panic!("Snapshot id of ref {ref_name} doesn't exist"))
332        })
333    }
334
335    /// Return all sort orders.
336    #[inline]
337    pub fn sort_orders_iter(&self) -> impl ExactSizeIterator<Item = &SortOrderRef> {
338        self.sort_orders.values()
339    }
340
341    /// Lookup sort order by id.
342    #[inline]
343    pub fn sort_order_by_id(&self, sort_order_id: i64) -> Option<&SortOrderRef> {
344        self.sort_orders.get(&sort_order_id)
345    }
346
347    /// Returns default sort order id.
348    #[inline]
349    pub fn default_sort_order(&self) -> &SortOrderRef {
350        self.sort_orders
351            .get(&self.default_sort_order_id)
352            .expect("Default order id has been set, but not found in table metadata!")
353    }
354
355    /// Returns default sort order id.
356    #[inline]
357    pub fn default_sort_order_id(&self) -> i64 {
358        self.default_sort_order_id
359    }
360
361    /// Returns properties of table.
362    #[inline]
363    pub fn properties(&self) -> &HashMap<String, String> {
364        &self.properties
365    }
366
367    /// Returns the base location for metadata files (manifests, manifest lists).
368    ///
369    /// Honors the `write.metadata.path` table property when set, otherwise defaults
370    /// to the `metadata` subdirectory under the table location.
371    pub fn metadata_location(&self) -> Result<String> {
372        Ok(self
373            .table_properties()?
374            .write_metadata_path
375            .unwrap_or_else(|| format!("{}/{}", self.location(), METADATA_FOLDER_NAME)))
376    }
377
378    /// Returns the metadata compression codec from table properties.
379    ///
380    /// Returns `CompressionCodec::None` if compression is disabled or not configured.
381    /// Returns `CompressionCodec::Gzip` if gzip compression is enabled.
382    ///
383    /// # Errors
384    ///
385    /// Returns an error if the compression codec property has an invalid value.
386    pub fn metadata_compression_codec(&self) -> Result<CompressionCodec> {
387        parse_metadata_file_compression(&self.properties)
388    }
389
390    /// Returns typed table properties parsed from the raw properties map with defaults.
391    pub fn table_properties(&self) -> Result<TableProperties> {
392        TableProperties::try_from(&self.properties).map_err(|e| {
393            Error::new(ErrorKind::DataInvalid, "Invalid table properties").with_source(e)
394        })
395    }
396
397    /// Return location of statistics files.
398    #[inline]
399    pub fn statistics_iter(&self) -> impl ExactSizeIterator<Item = &StatisticsFile> {
400        self.statistics.values()
401    }
402
403    /// Return location of partition statistics files.
404    #[inline]
405    pub fn partition_statistics_iter(
406        &self,
407    ) -> impl ExactSizeIterator<Item = &PartitionStatisticsFile> {
408        self.partition_statistics.values()
409    }
410
411    /// Get a statistics file for a snapshot id.
412    #[inline]
413    pub fn statistics_for_snapshot(&self, snapshot_id: i64) -> Option<&StatisticsFile> {
414        self.statistics.get(&snapshot_id)
415    }
416
417    /// Get a partition statistics file for a snapshot id.
418    #[inline]
419    pub fn partition_statistics_for_snapshot(
420        &self,
421        snapshot_id: i64,
422    ) -> Option<&PartitionStatisticsFile> {
423        self.partition_statistics.get(&snapshot_id)
424    }
425
426    fn construct_refs(&mut self) {
427        if let Some(current_snapshot_id) = self.current_snapshot_id
428            && !self.refs.contains_key(MAIN_BRANCH)
429        {
430            self.refs
431                .insert(MAIN_BRANCH.to_string(), SnapshotReference {
432                    snapshot_id: current_snapshot_id,
433                    retention: SnapshotRetention::Branch {
434                        min_snapshots_to_keep: None,
435                        max_snapshot_age_ms: None,
436                        max_ref_age_ms: None,
437                    },
438                });
439        }
440    }
441
442    /// Iterate over all encryption keys
443    #[inline]
444    pub fn encryption_keys_iter(&self) -> impl ExactSizeIterator<Item = &EncryptedKey> {
445        self.encryption_keys.values()
446    }
447
448    /// Get the encryption key for a given key id
449    #[inline]
450    pub fn encryption_key(&self, key_id: &str) -> Option<&EncryptedKey> {
451        self.encryption_keys.get(key_id)
452    }
453
454    /// Get the next row id to be assigned
455    #[inline]
456    pub fn next_row_id(&self) -> u64 {
457        self.next_row_id
458    }
459
460    /// Read table metadata from the given location.
461    pub async fn read_from(
462        file_io: &FileIO,
463        metadata_location: impl AsRef<str>,
464    ) -> Result<TableMetadata> {
465        let metadata_location = metadata_location.as_ref();
466        let input_file = file_io.new_input(metadata_location)?;
467        let metadata_content = input_file.read().await?;
468
469        // Check if the file is compressed by looking for the gzip "magic number".
470        let metadata = if metadata_content.len() > 2
471            && metadata_content[0] == 0x1F
472            && metadata_content[1] == 0x8B
473        {
474            let decompressed_data = CompressionCodec::gzip_default()
475                .decompress(metadata_content.to_vec())
476                .map_err(|e| {
477                    Error::new(
478                        ErrorKind::DataInvalid,
479                        "Trying to read compressed metadata file",
480                    )
481                    .with_context("file_path", metadata_location)
482                    .with_source(e)
483                })?;
484            serde_json::from_slice(&decompressed_data)?
485        } else {
486            serde_json::from_slice(&metadata_content)?
487        };
488
489        Ok(metadata)
490    }
491
492    /// Write table metadata to the given location.
493    pub async fn write_to(
494        &self,
495        file_io: &FileIO,
496        metadata_location: &MetadataLocation,
497    ) -> Result<()> {
498        let json_data = serde_json::to_vec(self)?;
499
500        // Check if compression codec from properties matches the one in metadata_location
501        let codec = parse_metadata_file_compression(&self.properties)?;
502
503        if codec != metadata_location.compression_codec() {
504            return Err(Error::new(
505                ErrorKind::DataInvalid,
506                format!(
507                    "Compression codec mismatch: metadata_location has {:?}, but table properties specify {:?}",
508                    metadata_location.compression_codec(),
509                    codec
510                ),
511            ));
512        }
513
514        // Apply compression based on codec
515        let data_to_write = match codec {
516            CompressionCodec::Gzip(_) => codec.compress(json_data)?,
517            CompressionCodec::None => json_data,
518            _ => {
519                return Err(Error::new(
520                    ErrorKind::DataInvalid,
521                    format!("Unsupported metadata compression codec: {codec:?}"),
522                ));
523            }
524        };
525
526        file_io
527            .new_output(metadata_location.to_string())?
528            .write(data_to_write.into())
529            .await
530    }
531
532    /// Normalize this partition spec.
533    ///
534    /// This is an internal method
535    /// meant to be called after constructing table metadata from untrusted sources.
536    /// We run this method after json deserialization.
537    /// All constructors for `TableMetadata` which are part of `iceberg-rust`
538    /// should return normalized `TableMetadata`.
539    pub(super) fn try_normalize(&mut self) -> Result<&mut Self> {
540        self.validate_current_schema()?;
541        self.normalize_current_snapshot()?;
542        self.construct_refs();
543        self.validate_refs()?;
544        self.validate_chronological_snapshot_logs()?;
545        self.validate_chronological_metadata_logs()?;
546        // Normalize location (remove trailing slash)
547        self.location = self.location.trim_end_matches('/').to_string();
548        self.validate_snapshot_sequence_number()?;
549        self.validate_schema_format_compatibility()?;
550        self.try_normalize_partition_spec()?;
551        self.try_normalize_sort_order()?;
552        Ok(self)
553    }
554
555    /// If the default partition spec is not present in specs, add it
556    fn try_normalize_partition_spec(&mut self) -> Result<()> {
557        if self
558            .partition_spec_by_id(self.default_spec.spec_id())
559            .is_none()
560        {
561            self.partition_specs.insert(
562                self.default_spec.spec_id(),
563                Arc::new(Arc::unwrap_or_clone(self.default_spec.clone())),
564            );
565        }
566
567        Ok(())
568    }
569
570    /// If the default sort order is unsorted but the sort order is not present, add it
571    fn try_normalize_sort_order(&mut self) -> Result<()> {
572        // Validate that sort order ID 0 (reserved for unsorted) has no fields
573        if let Some(sort_order) = self.sort_order_by_id(SortOrder::UNSORTED_ORDER_ID)
574            && !sort_order.fields.is_empty()
575        {
576            return Err(Error::new(
577                ErrorKind::Unexpected,
578                format!(
579                    "Sort order ID {} is reserved for unsorted order",
580                    SortOrder::UNSORTED_ORDER_ID
581                ),
582            ));
583        }
584
585        if self.sort_order_by_id(self.default_sort_order_id).is_some() {
586            return Ok(());
587        }
588
589        if self.default_sort_order_id != SortOrder::UNSORTED_ORDER_ID {
590            return Err(Error::new(
591                ErrorKind::DataInvalid,
592                format!(
593                    "No sort order exists with the default sort order id {}.",
594                    self.default_sort_order_id
595                ),
596            ));
597        }
598
599        let sort_order = SortOrder::unsorted_order();
600        self.sort_orders
601            .insert(SortOrder::UNSORTED_ORDER_ID, Arc::new(sort_order));
602        Ok(())
603    }
604
605    /// Validate the current schema is set and exists.
606    fn validate_current_schema(&self) -> Result<()> {
607        if self.schema_by_id(self.current_schema_id).is_none() {
608            return Err(Error::new(
609                ErrorKind::DataInvalid,
610                format!(
611                    "No schema exists with the current schema id {}.",
612                    self.current_schema_id
613                ),
614            ));
615        }
616        Ok(())
617    }
618
619    /// If current snapshot is Some(-1) then set it to None.
620    fn normalize_current_snapshot(&mut self) -> Result<()> {
621        if let Some(current_snapshot_id) = self.current_snapshot_id {
622            if current_snapshot_id == EMPTY_SNAPSHOT_ID {
623                self.current_snapshot_id = None;
624            } else if self.snapshot_by_id(current_snapshot_id).is_none() {
625                return Err(Error::new(
626                    ErrorKind::DataInvalid,
627                    format!(
628                        "Snapshot for current snapshot id {current_snapshot_id} does not exist in the existing snapshots list"
629                    ),
630                ));
631            }
632        }
633        Ok(())
634    }
635
636    /// Validate that all refs are valid (snapshot exists)
637    fn validate_refs(&self) -> Result<()> {
638        for (name, snapshot_ref) in self.refs.iter() {
639            if self.snapshot_by_id(snapshot_ref.snapshot_id).is_none() {
640                return Err(Error::new(
641                    ErrorKind::DataInvalid,
642                    format!(
643                        "Snapshot for reference {name} does not exist in the existing snapshots list"
644                    ),
645                ));
646            }
647        }
648
649        let main_ref = self.refs.get(MAIN_BRANCH);
650        if self.current_snapshot_id.is_some() {
651            if let Some(main_ref) = main_ref
652                && main_ref.snapshot_id != self.current_snapshot_id.unwrap_or_default()
653            {
654                return Err(Error::new(
655                    ErrorKind::DataInvalid,
656                    format!(
657                        "Current snapshot id does not match main branch ({:?} != {:?})",
658                        self.current_snapshot_id.unwrap_or_default(),
659                        main_ref.snapshot_id
660                    ),
661                ));
662            }
663        } else if main_ref.is_some() {
664            return Err(Error::new(
665                ErrorKind::DataInvalid,
666                "Current snapshot is not set, but main branch exists",
667            ));
668        }
669
670        Ok(())
671    }
672
673    /// Validate that for V1 Metadata the last_sequence_number is 0
674    fn validate_snapshot_sequence_number(&self) -> Result<()> {
675        if self.format_version < FormatVersion::V2 && self.last_sequence_number != 0 {
676            return Err(Error::new(
677                ErrorKind::DataInvalid,
678                format!(
679                    "Last sequence number must be 0 in v1. Found {}",
680                    self.last_sequence_number
681                ),
682            ));
683        }
684
685        if self.format_version >= FormatVersion::V2
686            && let Some(snapshot) = self
687                .snapshots
688                .values()
689                .find(|snapshot| snapshot.sequence_number() > self.last_sequence_number)
690        {
691            return Err(Error::new(
692                ErrorKind::DataInvalid,
693                format!(
694                    "Invalid snapshot with id {} and sequence number {} greater than last sequence number {}",
695                    snapshot.snapshot_id(),
696                    snapshot.sequence_number(),
697                    self.last_sequence_number
698                ),
699            ));
700        }
701
702        Ok(())
703    }
704
705    /// Validate snapshots logs are chronological and last updated is after the last snapshot log.
706    fn validate_chronological_snapshot_logs(&self) -> Result<()> {
707        for window in self.snapshot_log.windows(2) {
708            let (prev, curr) = (&window[0], &window[1]);
709            // commits can happen concurrently from different machines.
710            // A tolerance helps us avoid failure for small clock skew
711            if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
712                return Err(Error::new(
713                    ErrorKind::DataInvalid,
714                    "Expected sorted snapshot log entries",
715                ));
716            }
717        }
718
719        if let Some(last) = self.snapshot_log.last() {
720            // commits can happen concurrently from different machines.
721            // A tolerance helps us avoid failure for small clock skew
722            if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
723                return Err(Error::new(
724                    ErrorKind::DataInvalid,
725                    format!(
726                        "Invalid update timestamp {}: before last snapshot log entry at {}",
727                        self.last_updated_ms, last.timestamp_ms
728                    ),
729                ));
730            }
731        }
732        Ok(())
733    }
734
735    fn validate_chronological_metadata_logs(&self) -> Result<()> {
736        for window in self.metadata_log.windows(2) {
737            let (prev, curr) = (&window[0], &window[1]);
738            // commits can happen concurrently from different machines.
739            // A tolerance helps us avoid failure for small clock skew
740            if curr.timestamp_ms - prev.timestamp_ms < -ONE_MINUTE_MS {
741                return Err(Error::new(
742                    ErrorKind::DataInvalid,
743                    "Expected sorted metadata log entries",
744                ));
745            }
746        }
747
748        if let Some(last) = self.metadata_log.last() {
749            // commits can happen concurrently from different machines.
750            // A tolerance helps us avoid failure for small clock skew
751            if self.last_updated_ms - last.timestamp_ms < -ONE_MINUTE_MS {
752                return Err(Error::new(
753                    ErrorKind::DataInvalid,
754                    format!(
755                        "Invalid update timestamp {}: before last metadata log entry at {}",
756                        self.last_updated_ms, last.timestamp_ms
757                    ),
758                ));
759            }
760        }
761
762        Ok(())
763    }
764
765    /// Validates that every type used in the current schema is supported by the
766    /// table's format version.  Delegates to [`Schema::check_format_compatibility`].
767    fn validate_schema_format_compatibility(&self) -> Result<()> {
768        self.current_schema()
769            .check_format_compatibility(self.format_version)
770    }
771}
772
773pub(super) mod _serde {
774    use std::borrow::BorrowMut;
775    /// This is a helper module that defines types to help with serialization/deserialization.
776    /// For deserialization the input first gets read into either the [TableMetadataV1] or [TableMetadataV2] struct
777    /// and then converted into the [TableMetadata] struct. Serialization works the other way around.
778    /// [TableMetadataV1] and [TableMetadataV2] are internal struct that are only used for serialization and deserialization.
779    use std::collections::HashMap;
780    /// This is a helper module that defines types to help with serialization/deserialization.
781    /// For deserialization the input first gets read into either the [TableMetadataV1] or [TableMetadataV2] struct
782    /// and then converted into the [TableMetadata] struct. Serialization works the other way around.
783    /// [TableMetadataV1] and [TableMetadataV2] are internal struct that are only used for serialization and deserialization.
784    use std::sync::Arc;
785
786    use serde::{Deserialize, Serialize};
787    use uuid::Uuid;
788
789    use super::{
790        DEFAULT_PARTITION_SPEC_ID, EMPTY_SNAPSHOT_ID, FormatVersion, MAIN_BRANCH, MetadataLog,
791        SnapshotLog, TableMetadata,
792    };
793    use crate::spec::schema::_serde::{SchemaV1, SchemaV2};
794    use crate::spec::snapshot::_serde::{SnapshotV1, SnapshotV2, SnapshotV3};
795    use crate::spec::{
796        EncryptedKey, INITIAL_ROW_ID, PartitionField, PartitionSpec, PartitionSpecRef,
797        PartitionStatisticsFile, Schema, SchemaRef, Snapshot, SnapshotReference, SnapshotRetention,
798        SortOrder, StatisticsFile,
799    };
800    use crate::{Error, ErrorKind};
801
802    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
803    #[serde(untagged)]
804    pub(super) enum TableMetadataEnum {
805        V3(TableMetadataV3),
806        V2(TableMetadataV2),
807        V1(TableMetadataV1),
808    }
809
810    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
811    #[serde(rename_all = "kebab-case")]
812    /// Defines the structure of a v2 table metadata for serialization/deserialization
813    pub(super) struct TableMetadataV3 {
814        pub format_version: VersionNumber<3>,
815        #[serde(flatten)]
816        pub shared: TableMetadataV2V3Shared,
817        pub next_row_id: u64,
818        #[serde(skip_serializing_if = "Option::is_none")]
819        pub encryption_keys: Option<Vec<EncryptedKey>>,
820        #[serde(skip_serializing_if = "Option::is_none")]
821        pub snapshots: Option<Vec<SnapshotV3>>,
822    }
823
824    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
825    #[serde(rename_all = "kebab-case")]
826    /// Defines the structure of a v2 table metadata for serialization/deserialization
827    pub(super) struct TableMetadataV2V3Shared {
828        pub table_uuid: Uuid,
829        pub location: String,
830        pub last_sequence_number: i64,
831        pub last_updated_ms: i64,
832        pub last_column_id: i32,
833        pub schemas: Vec<SchemaV2>,
834        pub current_schema_id: i32,
835        pub partition_specs: Vec<PartitionSpec>,
836        pub default_spec_id: i32,
837        pub last_partition_id: i32,
838        #[serde(skip_serializing_if = "Option::is_none")]
839        pub properties: Option<HashMap<String, String>>,
840        #[serde(skip_serializing_if = "Option::is_none")]
841        pub current_snapshot_id: Option<i64>,
842        #[serde(skip_serializing_if = "Option::is_none")]
843        pub snapshot_log: Option<Vec<SnapshotLog>>,
844        #[serde(skip_serializing_if = "Option::is_none")]
845        pub metadata_log: Option<Vec<MetadataLog>>,
846        pub sort_orders: Vec<SortOrder>,
847        pub default_sort_order_id: i64,
848        #[serde(skip_serializing_if = "Option::is_none")]
849        pub refs: Option<HashMap<String, SnapshotReference>>,
850        #[serde(default, skip_serializing_if = "Vec::is_empty")]
851        pub statistics: Vec<StatisticsFile>,
852        #[serde(default, skip_serializing_if = "Vec::is_empty")]
853        pub partition_statistics: Vec<PartitionStatisticsFile>,
854    }
855
856    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
857    #[serde(rename_all = "kebab-case")]
858    /// Defines the structure of a v2 table metadata for serialization/deserialization
859    pub(super) struct TableMetadataV2 {
860        pub format_version: VersionNumber<2>,
861        #[serde(flatten)]
862        pub shared: TableMetadataV2V3Shared,
863        #[serde(skip_serializing_if = "Option::is_none")]
864        pub snapshots: Option<Vec<SnapshotV2>>,
865    }
866
867    #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
868    #[serde(rename_all = "kebab-case")]
869    /// Defines the structure of a v1 table metadata for serialization/deserialization
870    pub(super) struct TableMetadataV1 {
871        pub format_version: VersionNumber<1>,
872        #[serde(skip_serializing_if = "Option::is_none")]
873        pub table_uuid: Option<Uuid>,
874        pub location: String,
875        pub last_updated_ms: i64,
876        pub last_column_id: i32,
877        /// `schema` is optional to prioritize `schemas` and `current-schema-id`, allowing liberal reading of V1 metadata.
878        pub schema: Option<SchemaV1>,
879        #[serde(skip_serializing_if = "Option::is_none")]
880        pub schemas: Option<Vec<SchemaV1>>,
881        #[serde(skip_serializing_if = "Option::is_none")]
882        pub current_schema_id: Option<i32>,
883        /// `partition_spec` is optional to prioritize `partition_specs`, aligning with liberal reading of potentially invalid V1 metadata.
884        pub partition_spec: Option<Vec<PartitionField>>,
885        #[serde(skip_serializing_if = "Option::is_none")]
886        pub partition_specs: Option<Vec<PartitionSpec>>,
887        #[serde(skip_serializing_if = "Option::is_none")]
888        pub default_spec_id: Option<i32>,
889        #[serde(skip_serializing_if = "Option::is_none")]
890        pub last_partition_id: Option<i32>,
891        #[serde(skip_serializing_if = "Option::is_none")]
892        pub properties: Option<HashMap<String, String>>,
893        #[serde(skip_serializing_if = "Option::is_none")]
894        pub current_snapshot_id: Option<i64>,
895        #[serde(skip_serializing_if = "Option::is_none")]
896        pub snapshots: Option<Vec<SnapshotV1>>,
897        #[serde(skip_serializing_if = "Option::is_none")]
898        pub snapshot_log: Option<Vec<SnapshotLog>>,
899        #[serde(skip_serializing_if = "Option::is_none")]
900        pub metadata_log: Option<Vec<MetadataLog>>,
901        pub sort_orders: Option<Vec<SortOrder>>,
902        pub default_sort_order_id: Option<i64>,
903        #[serde(default, skip_serializing_if = "Vec::is_empty")]
904        pub statistics: Vec<StatisticsFile>,
905        #[serde(default, skip_serializing_if = "Vec::is_empty")]
906        pub partition_statistics: Vec<PartitionStatisticsFile>,
907    }
908
909    /// Helper to serialize and deserialize the format version.
910    #[derive(Debug, PartialEq, Eq)]
911    pub(crate) struct VersionNumber<const V: u8>;
912
913    impl Serialize for TableMetadata {
914        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
915        where S: serde::Serializer {
916            // we must do a clone here
917            let table_metadata_enum: TableMetadataEnum =
918                self.clone().try_into().map_err(serde::ser::Error::custom)?;
919
920            table_metadata_enum.serialize(serializer)
921        }
922    }
923
924    impl<const V: u8> Serialize for VersionNumber<V> {
925        fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
926        where S: serde::Serializer {
927            serializer.serialize_u8(V)
928        }
929    }
930
931    impl<'de, const V: u8> Deserialize<'de> for VersionNumber<V> {
932        fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
933        where D: serde::Deserializer<'de> {
934            let value = u8::deserialize(deserializer)?;
935            if value == V {
936                Ok(VersionNumber::<V>)
937            } else {
938                Err(serde::de::Error::custom("Invalid Version"))
939            }
940        }
941    }
942
943    impl TryFrom<TableMetadataEnum> for TableMetadata {
944        type Error = Error;
945        fn try_from(value: TableMetadataEnum) -> Result<Self, Error> {
946            match value {
947                TableMetadataEnum::V3(value) => value.try_into(),
948                TableMetadataEnum::V2(value) => value.try_into(),
949                TableMetadataEnum::V1(value) => value.try_into(),
950            }
951        }
952    }
953
954    impl TryFrom<TableMetadata> for TableMetadataEnum {
955        type Error = Error;
956        fn try_from(value: TableMetadata) -> Result<Self, Error> {
957            Ok(match value.format_version {
958                FormatVersion::V3 => TableMetadataEnum::V3(value.try_into()?),
959                FormatVersion::V2 => TableMetadataEnum::V2(value.into()),
960                FormatVersion::V1 => TableMetadataEnum::V1(value.try_into()?),
961            })
962        }
963    }
964
965    impl TryFrom<TableMetadataV3> for TableMetadata {
966        type Error = Error;
967        fn try_from(value: TableMetadataV3) -> Result<Self, Error> {
968            let TableMetadataV3 {
969                format_version: _,
970                shared: value,
971                next_row_id,
972                encryption_keys,
973                snapshots,
974            } = value;
975            let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
976                None
977            } else {
978                value.current_snapshot_id
979            };
980            let schemas = HashMap::from_iter(
981                value
982                    .schemas
983                    .into_iter()
984                    .map(|schema| Ok((schema.schema_id, Arc::new(schema.try_into()?))))
985                    .collect::<Result<Vec<_>, Error>>()?,
986            );
987
988            let current_schema: &SchemaRef =
989                schemas.get(&value.current_schema_id).ok_or_else(|| {
990                    Error::new(
991                        ErrorKind::DataInvalid,
992                        format!(
993                            "No schema exists with the current schema id {}.",
994                            value.current_schema_id
995                        ),
996                    )
997                })?;
998            let partition_specs = HashMap::from_iter(
999                value
1000                    .partition_specs
1001                    .into_iter()
1002                    .map(|x| (x.spec_id(), Arc::new(x))),
1003            );
1004            let default_spec_id = value.default_spec_id;
1005            let default_spec: PartitionSpecRef = partition_specs
1006                .get(&value.default_spec_id)
1007                .map(|spec| (**spec).clone())
1008                .or_else(|| {
1009                    (DEFAULT_PARTITION_SPEC_ID == default_spec_id)
1010                        .then(PartitionSpec::unpartition_spec)
1011                })
1012                .ok_or_else(|| {
1013                    Error::new(
1014                        ErrorKind::DataInvalid,
1015                        format!("Default partition spec {default_spec_id} not found"),
1016                    )
1017                })?
1018                .into();
1019            let default_partition_type = default_spec.partition_type(current_schema)?;
1020
1021            let mut metadata = TableMetadata {
1022                format_version: FormatVersion::V3,
1023                table_uuid: value.table_uuid,
1024                location: value.location,
1025                last_sequence_number: value.last_sequence_number,
1026                last_updated_ms: value.last_updated_ms,
1027                last_column_id: value.last_column_id,
1028                current_schema_id: value.current_schema_id,
1029                schemas,
1030                partition_specs,
1031                default_partition_type,
1032                default_spec,
1033                last_partition_id: value.last_partition_id,
1034                properties: value.properties.unwrap_or_default(),
1035                current_snapshot_id,
1036                snapshots: snapshots
1037                    .map(|snapshots| {
1038                        HashMap::from_iter(
1039                            snapshots
1040                                .into_iter()
1041                                .map(|x| (x.snapshot_id, Arc::new(x.into()))),
1042                        )
1043                    })
1044                    .unwrap_or_default(),
1045                snapshot_log: value.snapshot_log.unwrap_or_default(),
1046                metadata_log: value.metadata_log.unwrap_or_default(),
1047                sort_orders: HashMap::from_iter(
1048                    value
1049                        .sort_orders
1050                        .into_iter()
1051                        .map(|x| (x.order_id, Arc::new(x))),
1052                ),
1053                default_sort_order_id: value.default_sort_order_id,
1054                refs: value.refs.unwrap_or_else(|| {
1055                    if let Some(snapshot_id) = current_snapshot_id {
1056                        HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
1057                            snapshot_id,
1058                            retention: SnapshotRetention::Branch {
1059                                min_snapshots_to_keep: None,
1060                                max_snapshot_age_ms: None,
1061                                max_ref_age_ms: None,
1062                            },
1063                        })])
1064                    } else {
1065                        HashMap::new()
1066                    }
1067                }),
1068                statistics: index_statistics(value.statistics),
1069                partition_statistics: index_partition_statistics(value.partition_statistics),
1070                encryption_keys: encryption_keys
1071                    .map(|keys| {
1072                        HashMap::from_iter(keys.into_iter().map(|key| (key.key_id.clone(), key)))
1073                    })
1074                    .unwrap_or_default(),
1075                next_row_id,
1076            };
1077
1078            metadata.borrow_mut().try_normalize()?;
1079            Ok(metadata)
1080        }
1081    }
1082
1083    impl TryFrom<TableMetadataV2> for TableMetadata {
1084        type Error = Error;
1085        fn try_from(value: TableMetadataV2) -> Result<Self, Error> {
1086            let snapshots = value.snapshots;
1087            let value = value.shared;
1088            let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
1089                None
1090            } else {
1091                value.current_snapshot_id
1092            };
1093            let schemas = HashMap::from_iter(
1094                value
1095                    .schemas
1096                    .into_iter()
1097                    .map(|schema| Ok((schema.schema_id, Arc::new(schema.try_into()?))))
1098                    .collect::<Result<Vec<_>, Error>>()?,
1099            );
1100
1101            let current_schema: &SchemaRef =
1102                schemas.get(&value.current_schema_id).ok_or_else(|| {
1103                    Error::new(
1104                        ErrorKind::DataInvalid,
1105                        format!(
1106                            "No schema exists with the current schema id {}.",
1107                            value.current_schema_id
1108                        ),
1109                    )
1110                })?;
1111            let partition_specs = HashMap::from_iter(
1112                value
1113                    .partition_specs
1114                    .into_iter()
1115                    .map(|x| (x.spec_id(), Arc::new(x))),
1116            );
1117            let default_spec_id = value.default_spec_id;
1118            let default_spec: PartitionSpecRef = partition_specs
1119                .get(&value.default_spec_id)
1120                .map(|spec| (**spec).clone())
1121                .or_else(|| {
1122                    (DEFAULT_PARTITION_SPEC_ID == default_spec_id)
1123                        .then(PartitionSpec::unpartition_spec)
1124                })
1125                .ok_or_else(|| {
1126                    Error::new(
1127                        ErrorKind::DataInvalid,
1128                        format!("Default partition spec {default_spec_id} not found"),
1129                    )
1130                })?
1131                .into();
1132            let default_partition_type = default_spec.partition_type(current_schema)?;
1133
1134            let mut metadata = TableMetadata {
1135                format_version: FormatVersion::V2,
1136                table_uuid: value.table_uuid,
1137                location: value.location,
1138                last_sequence_number: value.last_sequence_number,
1139                last_updated_ms: value.last_updated_ms,
1140                last_column_id: value.last_column_id,
1141                current_schema_id: value.current_schema_id,
1142                schemas,
1143                partition_specs,
1144                default_partition_type,
1145                default_spec,
1146                last_partition_id: value.last_partition_id,
1147                properties: value.properties.unwrap_or_default(),
1148                current_snapshot_id,
1149                snapshots: snapshots
1150                    .map(|snapshots| {
1151                        HashMap::from_iter(
1152                            snapshots
1153                                .into_iter()
1154                                .map(|x| (x.snapshot_id, Arc::new(x.into()))),
1155                        )
1156                    })
1157                    .unwrap_or_default(),
1158                snapshot_log: value.snapshot_log.unwrap_or_default(),
1159                metadata_log: value.metadata_log.unwrap_or_default(),
1160                sort_orders: HashMap::from_iter(
1161                    value
1162                        .sort_orders
1163                        .into_iter()
1164                        .map(|x| (x.order_id, Arc::new(x))),
1165                ),
1166                default_sort_order_id: value.default_sort_order_id,
1167                refs: value.refs.unwrap_or_else(|| {
1168                    if let Some(snapshot_id) = current_snapshot_id {
1169                        HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
1170                            snapshot_id,
1171                            retention: SnapshotRetention::Branch {
1172                                min_snapshots_to_keep: None,
1173                                max_snapshot_age_ms: None,
1174                                max_ref_age_ms: None,
1175                            },
1176                        })])
1177                    } else {
1178                        HashMap::new()
1179                    }
1180                }),
1181                statistics: index_statistics(value.statistics),
1182                partition_statistics: index_partition_statistics(value.partition_statistics),
1183                encryption_keys: HashMap::new(),
1184                next_row_id: INITIAL_ROW_ID,
1185            };
1186
1187            metadata.borrow_mut().try_normalize()?;
1188            Ok(metadata)
1189        }
1190    }
1191
1192    impl TryFrom<TableMetadataV1> for TableMetadata {
1193        type Error = Error;
1194        fn try_from(value: TableMetadataV1) -> Result<Self, Error> {
1195            let current_snapshot_id = if value.current_snapshot_id == Some(EMPTY_SNAPSHOT_ID) {
1196                None
1197            } else {
1198                value.current_snapshot_id
1199            };
1200
1201            let (schemas, current_schema_id, current_schema) =
1202                if let (Some(schemas_vec), Some(schema_id)) =
1203                    (&value.schemas, value.current_schema_id)
1204                {
1205                    // Option 1: Use 'schemas' + 'current_schema_id'
1206                    let schema_map = HashMap::from_iter(
1207                        schemas_vec
1208                            .clone()
1209                            .into_iter()
1210                            .map(|schema| {
1211                                let schema: Schema = schema.try_into()?;
1212                                Ok((schema.schema_id(), Arc::new(schema)))
1213                            })
1214                            .collect::<Result<Vec<_>, Error>>()?,
1215                    );
1216
1217                    let schema = schema_map
1218                        .get(&schema_id)
1219                        .ok_or_else(|| {
1220                            Error::new(
1221                                ErrorKind::DataInvalid,
1222                                format!("No schema exists with the current schema id {schema_id}."),
1223                            )
1224                        })?
1225                        .clone();
1226                    (schema_map, schema_id, schema)
1227                } else if let Some(schema) = value.schema {
1228                    // Option 2: Fall back to `schema`
1229                    let schema: Schema = schema.try_into()?;
1230                    let schema_id = schema.schema_id();
1231                    let schema_arc = Arc::new(schema);
1232                    let schema_map = HashMap::from_iter(vec![(schema_id, schema_arc.clone())]);
1233                    (schema_map, schema_id, schema_arc)
1234                } else {
1235                    // Option 3: No valid schema configuration found
1236                    return Err(Error::new(
1237                        ErrorKind::DataInvalid,
1238                        "No valid schema configuration found in table metadata",
1239                    ));
1240                };
1241
1242            // Prioritize 'partition_specs' over 'partition_spec'
1243            let partition_specs = if let Some(specs_vec) = value.partition_specs {
1244                // Option 1: Use 'partition_specs'
1245                specs_vec
1246                    .into_iter()
1247                    .map(|x| (x.spec_id(), Arc::new(x)))
1248                    .collect::<HashMap<_, _>>()
1249            } else if let Some(partition_spec) = value.partition_spec {
1250                // Option 2: Fall back to 'partition_spec'
1251                let spec = PartitionSpec::builder(current_schema.clone())
1252                    .with_spec_id(DEFAULT_PARTITION_SPEC_ID)
1253                    .add_unbound_fields(partition_spec.into_iter().map(|f| f.into_unbound()))?
1254                    .build()?;
1255
1256                HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))])
1257            } else {
1258                // Option 3: Create empty partition spec
1259                let spec = PartitionSpec::builder(current_schema.clone())
1260                    .with_spec_id(DEFAULT_PARTITION_SPEC_ID)
1261                    .build()?;
1262
1263                HashMap::from_iter(vec![(DEFAULT_PARTITION_SPEC_ID, Arc::new(spec))])
1264            };
1265
1266            // Get the default_spec_id, prioritizing the explicit value if provided
1267            let default_spec_id = value
1268                .default_spec_id
1269                .unwrap_or_else(|| partition_specs.keys().copied().max().unwrap_or_default());
1270
1271            // Get the default spec
1272            let default_spec: PartitionSpecRef = partition_specs
1273                .get(&default_spec_id)
1274                .map(|x| Arc::unwrap_or_clone(x.clone()))
1275                .ok_or_else(|| {
1276                    Error::new(
1277                        ErrorKind::DataInvalid,
1278                        format!("Default partition spec {default_spec_id} not found"),
1279                    )
1280                })?
1281                .into();
1282            let default_partition_type = default_spec.partition_type(&current_schema)?;
1283
1284            let mut metadata = TableMetadata {
1285                format_version: FormatVersion::V1,
1286                table_uuid: value.table_uuid.unwrap_or_default(),
1287                location: value.location,
1288                last_sequence_number: 0,
1289                last_updated_ms: value.last_updated_ms,
1290                last_column_id: value.last_column_id,
1291                current_schema_id,
1292                default_spec,
1293                default_partition_type,
1294                last_partition_id: value
1295                    .last_partition_id
1296                    .unwrap_or_else(|| partition_specs.keys().copied().max().unwrap_or_default()),
1297                partition_specs,
1298                schemas,
1299                properties: value.properties.unwrap_or_default(),
1300                current_snapshot_id,
1301                snapshots: value
1302                    .snapshots
1303                    .map(|snapshots| {
1304                        Ok::<_, Error>(HashMap::from_iter(
1305                            snapshots
1306                                .into_iter()
1307                                .map(|x| Ok((x.snapshot_id, Arc::new(x.try_into()?))))
1308                                .collect::<Result<Vec<_>, Error>>()?,
1309                        ))
1310                    })
1311                    .transpose()?
1312                    .unwrap_or_default(),
1313                snapshot_log: value.snapshot_log.unwrap_or_default(),
1314                metadata_log: value.metadata_log.unwrap_or_default(),
1315                sort_orders: match value.sort_orders {
1316                    Some(sort_orders) => HashMap::from_iter(
1317                        sort_orders.into_iter().map(|x| (x.order_id, Arc::new(x))),
1318                    ),
1319                    None => HashMap::new(),
1320                },
1321                default_sort_order_id: value
1322                    .default_sort_order_id
1323                    .unwrap_or(SortOrder::UNSORTED_ORDER_ID),
1324                refs: if let Some(snapshot_id) = current_snapshot_id {
1325                    HashMap::from_iter(vec![(MAIN_BRANCH.to_string(), SnapshotReference {
1326                        snapshot_id,
1327                        retention: SnapshotRetention::Branch {
1328                            min_snapshots_to_keep: None,
1329                            max_snapshot_age_ms: None,
1330                            max_ref_age_ms: None,
1331                        },
1332                    })])
1333                } else {
1334                    HashMap::new()
1335                },
1336                statistics: index_statistics(value.statistics),
1337                partition_statistics: index_partition_statistics(value.partition_statistics),
1338                encryption_keys: HashMap::new(),
1339                next_row_id: INITIAL_ROW_ID, // v1 has no row lineage
1340            };
1341
1342            metadata.borrow_mut().try_normalize()?;
1343            Ok(metadata)
1344        }
1345    }
1346
1347    impl TryFrom<TableMetadata> for TableMetadataV3 {
1348        type Error = Error;
1349
1350        fn try_from(mut v: TableMetadata) -> Result<Self, Self::Error> {
1351            let next_row_id = v.next_row_id;
1352            let encryption_keys = std::mem::take(&mut v.encryption_keys);
1353            let snapshots = std::mem::take(&mut v.snapshots);
1354            let shared = v.into();
1355
1356            Ok(TableMetadataV3 {
1357                format_version: VersionNumber::<3>,
1358                shared,
1359                next_row_id,
1360                encryption_keys: if encryption_keys.is_empty() {
1361                    None
1362                } else {
1363                    Some(encryption_keys.into_values().collect())
1364                },
1365                snapshots: if snapshots.is_empty() {
1366                    None
1367                } else {
1368                    Some(
1369                        snapshots
1370                            .into_values()
1371                            .map(|s| SnapshotV3::try_from(Arc::unwrap_or_clone(s)))
1372                            .collect::<Result<_, _>>()?,
1373                    )
1374                },
1375            })
1376        }
1377    }
1378
1379    impl From<TableMetadata> for TableMetadataV2 {
1380        fn from(mut v: TableMetadata) -> Self {
1381            let snapshots = std::mem::take(&mut v.snapshots);
1382            let shared = v.into();
1383
1384            TableMetadataV2 {
1385                format_version: VersionNumber::<2>,
1386                shared,
1387                snapshots: if snapshots.is_empty() {
1388                    None
1389                } else {
1390                    Some(
1391                        snapshots
1392                            .into_values()
1393                            .map(|s| SnapshotV2::from(Arc::unwrap_or_clone(s)))
1394                            .collect(),
1395                    )
1396                },
1397            }
1398        }
1399    }
1400
1401    impl From<TableMetadata> for TableMetadataV2V3Shared {
1402        fn from(v: TableMetadata) -> Self {
1403            TableMetadataV2V3Shared {
1404                table_uuid: v.table_uuid,
1405                location: v.location,
1406                last_sequence_number: v.last_sequence_number,
1407                last_updated_ms: v.last_updated_ms,
1408                last_column_id: v.last_column_id,
1409                schemas: v
1410                    .schemas
1411                    .into_values()
1412                    .map(|x| {
1413                        Arc::try_unwrap(x)
1414                            .unwrap_or_else(|schema| schema.as_ref().clone())
1415                            .into()
1416                    })
1417                    .collect(),
1418                current_schema_id: v.current_schema_id,
1419                partition_specs: v
1420                    .partition_specs
1421                    .into_values()
1422                    .map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
1423                    .collect(),
1424                default_spec_id: v.default_spec.spec_id(),
1425                last_partition_id: v.last_partition_id,
1426                properties: if v.properties.is_empty() {
1427                    None
1428                } else {
1429                    Some(v.properties)
1430                },
1431                current_snapshot_id: v.current_snapshot_id,
1432                snapshot_log: if v.snapshot_log.is_empty() {
1433                    None
1434                } else {
1435                    Some(v.snapshot_log)
1436                },
1437                metadata_log: if v.metadata_log.is_empty() {
1438                    None
1439                } else {
1440                    Some(v.metadata_log)
1441                },
1442                sort_orders: v
1443                    .sort_orders
1444                    .into_values()
1445                    .map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
1446                    .collect(),
1447                default_sort_order_id: v.default_sort_order_id,
1448                refs: Some(v.refs),
1449                statistics: v.statistics.into_values().collect(),
1450                partition_statistics: v.partition_statistics.into_values().collect(),
1451            }
1452        }
1453    }
1454
1455    impl TryFrom<TableMetadata> for TableMetadataV1 {
1456        type Error = Error;
1457        fn try_from(v: TableMetadata) -> Result<Self, Error> {
1458            Ok(TableMetadataV1 {
1459                format_version: VersionNumber::<1>,
1460                table_uuid: Some(v.table_uuid),
1461                location: v.location,
1462                last_updated_ms: v.last_updated_ms,
1463                last_column_id: v.last_column_id,
1464                schema: Some(
1465                    v.schemas
1466                        .get(&v.current_schema_id)
1467                        .ok_or(Error::new(
1468                            ErrorKind::Unexpected,
1469                            "current_schema_id not found in schemas",
1470                        ))?
1471                        .as_ref()
1472                        .clone()
1473                        .into(),
1474                ),
1475                schemas: Some(
1476                    v.schemas
1477                        .into_values()
1478                        .map(|x| {
1479                            Arc::try_unwrap(x)
1480                                .unwrap_or_else(|schema| schema.as_ref().clone())
1481                                .into()
1482                        })
1483                        .collect(),
1484                ),
1485                current_schema_id: Some(v.current_schema_id),
1486                partition_spec: Some(v.default_spec.fields().to_vec()),
1487                partition_specs: Some(
1488                    v.partition_specs
1489                        .into_values()
1490                        .map(|x| Arc::try_unwrap(x).unwrap_or_else(|s| s.as_ref().clone()))
1491                        .collect(),
1492                ),
1493                default_spec_id: Some(v.default_spec.spec_id()),
1494                last_partition_id: Some(v.last_partition_id),
1495                properties: if v.properties.is_empty() {
1496                    None
1497                } else {
1498                    Some(v.properties)
1499                },
1500                current_snapshot_id: v.current_snapshot_id,
1501                snapshots: if v.snapshots.is_empty() {
1502                    None
1503                } else {
1504                    Some(
1505                        v.snapshots
1506                            .into_values()
1507                            .map(|x| Snapshot::clone(&x).into())
1508                            .collect(),
1509                    )
1510                },
1511                snapshot_log: if v.snapshot_log.is_empty() {
1512                    None
1513                } else {
1514                    Some(v.snapshot_log)
1515                },
1516                metadata_log: if v.metadata_log.is_empty() {
1517                    None
1518                } else {
1519                    Some(v.metadata_log)
1520                },
1521                sort_orders: Some(
1522                    v.sort_orders
1523                        .into_values()
1524                        .map(|s| Arc::try_unwrap(s).unwrap_or_else(|s| s.as_ref().clone()))
1525                        .collect(),
1526                ),
1527                default_sort_order_id: Some(v.default_sort_order_id),
1528                statistics: v.statistics.into_values().collect(),
1529                partition_statistics: v.partition_statistics.into_values().collect(),
1530            })
1531        }
1532    }
1533
1534    fn index_statistics(statistics: Vec<StatisticsFile>) -> HashMap<i64, StatisticsFile> {
1535        statistics
1536            .into_iter()
1537            .rev()
1538            .map(|s| (s.snapshot_id, s))
1539            .collect()
1540    }
1541
1542    fn index_partition_statistics(
1543        statistics: Vec<PartitionStatisticsFile>,
1544    ) -> HashMap<i64, PartitionStatisticsFile> {
1545        statistics
1546            .into_iter()
1547            .rev()
1548            .map(|s| (s.snapshot_id, s))
1549            .collect()
1550    }
1551}
1552
1553#[derive(Debug, Serialize_repr, Deserialize_repr, PartialEq, Eq, Clone, Copy, Hash)]
1554#[repr(u8)]
1555/// Iceberg format version
1556pub enum FormatVersion {
1557    /// Iceberg spec version 1
1558    V1 = 1u8,
1559    /// Iceberg spec version 2
1560    V2 = 2u8,
1561    /// Iceberg spec version 3
1562    V3 = 3u8,
1563}
1564
1565impl PartialOrd for FormatVersion {
1566    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
1567        Some(self.cmp(other))
1568    }
1569}
1570
1571impl Ord for FormatVersion {
1572    fn cmp(&self, other: &Self) -> Ordering {
1573        (*self as u8).cmp(&(*other as u8))
1574    }
1575}
1576
1577impl Display for FormatVersion {
1578    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
1579        match self {
1580            FormatVersion::V1 => write!(f, "v1"),
1581            FormatVersion::V2 => write!(f, "v2"),
1582            FormatVersion::V3 => write!(f, "v3"),
1583        }
1584    }
1585}
1586
1587#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
1588#[serde(rename_all = "kebab-case")]
1589/// Encodes changes to the previous metadata files for the table
1590pub struct MetadataLog {
1591    /// The file for the log.
1592    pub metadata_file: String,
1593    /// Time new metadata was created
1594    pub timestamp_ms: i64,
1595}
1596
1597#[derive(Debug, Serialize, Deserialize, PartialEq, Eq, Clone)]
1598#[serde(rename_all = "kebab-case")]
1599/// A log of when each snapshot was made.
1600pub struct SnapshotLog {
1601    /// Id of the snapshot.
1602    pub snapshot_id: i64,
1603    /// Last updated timestamp
1604    pub timestamp_ms: i64,
1605}
1606
1607impl SnapshotLog {
1608    /// Returns the last updated timestamp as a DateTime<Utc> with millisecond precision
1609    pub fn timestamp(self) -> Result<DateTime<Utc>> {
1610        timestamp_ms_to_utc(self.timestamp_ms)
1611    }
1612
1613    /// Returns the timestamp in milliseconds
1614    #[inline]
1615    pub fn timestamp_ms(&self) -> i64 {
1616        self.timestamp_ms
1617    }
1618}
1619
1620#[cfg(test)]
1621mod tests {
1622    use std::collections::HashMap;
1623    use std::fs;
1624    use std::sync::Arc;
1625
1626    use anyhow::Result;
1627    use base64::Engine as _;
1628    use pretty_assertions::assert_eq;
1629    use tempfile::TempDir;
1630    use uuid::Uuid;
1631
1632    use super::{FormatVersion, MetadataLog, SnapshotLog, TableMetadataBuilder};
1633    use crate::catalog::MetadataLocation;
1634    use crate::compression::CompressionCodec;
1635    use crate::io::FileIO;
1636    use crate::spec::table_metadata::TableMetadata;
1637    use crate::spec::{
1638        BlobMetadata, EncryptedKey, INITIAL_ROW_ID, Literal, NestedField, NullOrder, Operation,
1639        PartitionSpec, PartitionStatisticsFile, PrimitiveLiteral, PrimitiveType, Schema, Snapshot,
1640        SnapshotReference, SnapshotRetention, SortDirection, SortField, SortOrder, StatisticsFile,
1641        Summary, TableProperties, Transform, Type, UnboundPartitionField,
1642    };
1643    use crate::{ErrorKind, TableCreation};
1644
1645    fn check_table_metadata_serde(json: &str, expected_type: TableMetadata) {
1646        let desered_type: TableMetadata = serde_json::from_str(json).unwrap();
1647        assert_eq!(desered_type, expected_type);
1648
1649        let sered_json = serde_json::to_string(&expected_type).unwrap();
1650        let parsed_json_value = serde_json::from_str::<TableMetadata>(&sered_json).unwrap();
1651
1652        assert_eq!(parsed_json_value, desered_type);
1653    }
1654
1655    fn get_test_table_metadata(file_name: &str) -> TableMetadata {
1656        let path = format!("testdata/table_metadata/{file_name}");
1657        let metadata: String = fs::read_to_string(path).unwrap();
1658
1659        serde_json::from_str(&metadata).unwrap()
1660    }
1661
1662    /// Loads a test table metadata and relocates it to `location`, so that derived
1663    /// metadata paths point at a writable (e.g. temp) directory.
1664    fn get_test_table_metadata_at(file_name: &str, location: &str) -> TableMetadata {
1665        TableMetadataBuilder::new_from_metadata(get_test_table_metadata(file_name), None)
1666            .set_location(location.to_string())
1667            .build()
1668            .unwrap()
1669            .metadata
1670    }
1671
1672    #[test]
1673    fn test_table_data_v2() {
1674        let data = r#"
1675            {
1676                "format-version" : 2,
1677                "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
1678                "location": "s3://b/wh/data.db/table",
1679                "last-sequence-number" : 1,
1680                "last-updated-ms": 1515100955770,
1681                "last-column-id": 1,
1682                "schemas": [
1683                    {
1684                        "schema-id" : 1,
1685                        "type" : "struct",
1686                        "fields" :[
1687                            {
1688                                "id": 1,
1689                                "name": "struct_name",
1690                                "required": true,
1691                                "type": "fixed[1]"
1692                            },
1693                            {
1694                                "id": 4,
1695                                "name": "ts",
1696                                "required": true,
1697                                "type": "timestamp"
1698                            }
1699                        ]
1700                    }
1701                ],
1702                "current-schema-id" : 1,
1703                "partition-specs": [
1704                    {
1705                        "spec-id": 0,
1706                        "fields": [
1707                            {
1708                                "source-id": 4,
1709                                "field-id": 1000,
1710                                "name": "ts_day",
1711                                "transform": "day"
1712                            }
1713                        ]
1714                    }
1715                ],
1716                "default-spec-id": 0,
1717                "last-partition-id": 1000,
1718                "properties": {
1719                    "commit.retry.num-retries": "1"
1720                },
1721                "metadata-log": [
1722                    {
1723                        "metadata-file": "s3://bucket/.../v1.json",
1724                        "timestamp-ms": 1515100
1725                    }
1726                ],
1727                "refs": {},
1728                "sort-orders": [
1729                    {
1730                    "order-id": 0,
1731                    "fields": []
1732                    }
1733                ],
1734                "default-sort-order-id": 0
1735            }
1736        "#;
1737
1738        let schema = Schema::builder()
1739            .with_schema_id(1)
1740            .with_fields(vec![
1741                Arc::new(NestedField::required(
1742                    1,
1743                    "struct_name",
1744                    Type::Primitive(PrimitiveType::Fixed(1)),
1745                )),
1746                Arc::new(NestedField::required(
1747                    4,
1748                    "ts",
1749                    Type::Primitive(PrimitiveType::Timestamp),
1750                )),
1751            ])
1752            .build()
1753            .unwrap();
1754
1755        let partition_spec = PartitionSpec::builder(schema.clone())
1756            .with_spec_id(0)
1757            .add_unbound_field(UnboundPartitionField {
1758                name: "ts_day".to_string(),
1759                transform: Transform::Day,
1760                source_id: 4,
1761                field_id: Some(1000),
1762            })
1763            .unwrap()
1764            .build()
1765            .unwrap();
1766
1767        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
1768        let expected = TableMetadata {
1769            format_version: FormatVersion::V2,
1770            table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
1771            location: "s3://b/wh/data.db/table".to_string(),
1772            last_updated_ms: 1515100955770,
1773            last_column_id: 1,
1774            schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
1775            current_schema_id: 1,
1776            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
1777            default_partition_type,
1778            default_spec: partition_spec.into(),
1779            last_partition_id: 1000,
1780            default_sort_order_id: 0,
1781            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
1782            snapshots: HashMap::default(),
1783            current_snapshot_id: None,
1784            last_sequence_number: 1,
1785            properties: HashMap::from_iter(vec![(
1786                "commit.retry.num-retries".to_string(),
1787                "1".to_string(),
1788            )]),
1789            snapshot_log: Vec::new(),
1790            metadata_log: vec![MetadataLog {
1791                metadata_file: "s3://bucket/.../v1.json".to_string(),
1792                timestamp_ms: 1515100,
1793            }],
1794            refs: HashMap::new(),
1795            statistics: HashMap::new(),
1796            partition_statistics: HashMap::new(),
1797            encryption_keys: HashMap::new(),
1798            next_row_id: INITIAL_ROW_ID,
1799        };
1800
1801        let expected_json_value = serde_json::to_value(&expected).unwrap();
1802        check_table_metadata_serde(data, expected);
1803
1804        let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
1805        assert_eq!(json_value, expected_json_value);
1806    }
1807
1808    #[test]
1809    fn test_table_data_v3() {
1810        let data = r#"
1811            {
1812                "format-version" : 3,
1813                "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
1814                "location": "s3://b/wh/data.db/table",
1815                "last-sequence-number" : 1,
1816                "last-updated-ms": 1515100955770,
1817                "last-column-id": 1,
1818                "next-row-id": 5,
1819                "schemas": [
1820                    {
1821                        "schema-id" : 1,
1822                        "type" : "struct",
1823                        "fields" :[
1824                            {
1825                                "id": 4,
1826                                "name": "ts",
1827                                "required": true,
1828                                "type": "timestamp"
1829                            }
1830                        ]
1831                    }
1832                ],
1833                "current-schema-id" : 1,
1834                "partition-specs": [
1835                    {
1836                        "spec-id": 0,
1837                        "fields": [
1838                            {
1839                                "source-id": 4,
1840                                "field-id": 1000,
1841                                "name": "ts_day",
1842                                "transform": "day"
1843                            }
1844                        ]
1845                    }
1846                ],
1847                "default-spec-id": 0,
1848                "last-partition-id": 1000,
1849                "properties": {
1850                    "commit.retry.num-retries": "1"
1851                },
1852                "metadata-log": [
1853                    {
1854                        "metadata-file": "s3://bucket/.../v1.json",
1855                        "timestamp-ms": 1515100
1856                    }
1857                ],
1858                "refs": {},
1859                "snapshots" : [ {
1860                    "snapshot-id" : 1,
1861                    "timestamp-ms" : 1662532818843,
1862                    "sequence-number" : 0,
1863                    "first-row-id" : 0,
1864                    "added-rows" : 4,
1865                    "key-id" : "key1",
1866                    "summary" : {
1867                        "operation" : "append"
1868                    },
1869                    "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
1870                    "schema-id" : 0
1871                    }
1872                ],
1873                "encryption-keys": [
1874                    {
1875                        "key-id": "key1",
1876                        "encrypted-by-id": "KMS",
1877                        "encrypted-key-metadata": "c29tZS1lbmNyeXB0aW9uLWtleQ==",
1878                        "properties": {
1879                            "p1": "v1"
1880                        }
1881                    }
1882                ],
1883                "sort-orders": [
1884                    {
1885                    "order-id": 0,
1886                    "fields": []
1887                    }
1888                ],
1889                "default-sort-order-id": 0
1890            }
1891        "#;
1892
1893        let schema = Schema::builder()
1894            .with_schema_id(1)
1895            .with_fields(vec![Arc::new(NestedField::required(
1896                4,
1897                "ts",
1898                Type::Primitive(PrimitiveType::Timestamp),
1899            ))])
1900            .build()
1901            .unwrap();
1902
1903        let partition_spec = PartitionSpec::builder(schema.clone())
1904            .with_spec_id(0)
1905            .add_unbound_field(UnboundPartitionField {
1906                name: "ts_day".to_string(),
1907                transform: Transform::Day,
1908                source_id: 4,
1909                field_id: Some(1000),
1910            })
1911            .unwrap()
1912            .build()
1913            .unwrap();
1914
1915        let snapshot = Snapshot::builder()
1916            .with_snapshot_id(1)
1917            .with_timestamp_ms(1662532818843)
1918            .with_sequence_number(0)
1919            .with_row_range(0, 4)
1920            .with_encryption_key_id(Some("key1".to_string()))
1921            .with_summary(Summary {
1922                operation: Operation::Append,
1923                additional_properties: HashMap::new(),
1924            })
1925            .with_manifest_list("/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro".to_string())
1926            .with_schema_id(0)
1927            .build();
1928
1929        let encryption_key = EncryptedKey::builder()
1930            .key_id("key1".to_string())
1931            .encrypted_by_id("KMS".to_string())
1932            .encrypted_key_metadata(
1933                base64::prelude::BASE64_STANDARD
1934                    .decode("c29tZS1lbmNyeXB0aW9uLWtleQ==")
1935                    .unwrap(),
1936            )
1937            .properties(HashMap::from_iter(vec![(
1938                "p1".to_string(),
1939                "v1".to_string(),
1940            )]))
1941            .build();
1942
1943        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
1944        let expected = TableMetadata {
1945            format_version: FormatVersion::V3,
1946            table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
1947            location: "s3://b/wh/data.db/table".to_string(),
1948            last_updated_ms: 1515100955770,
1949            last_column_id: 1,
1950            schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
1951            current_schema_id: 1,
1952            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
1953            default_partition_type,
1954            default_spec: partition_spec.into(),
1955            last_partition_id: 1000,
1956            default_sort_order_id: 0,
1957            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
1958            snapshots: HashMap::from_iter(vec![(1, snapshot.into())]),
1959            current_snapshot_id: None,
1960            last_sequence_number: 1,
1961            properties: HashMap::from_iter(vec![(
1962                "commit.retry.num-retries".to_string(),
1963                "1".to_string(),
1964            )]),
1965            snapshot_log: Vec::new(),
1966            metadata_log: vec![MetadataLog {
1967                metadata_file: "s3://bucket/.../v1.json".to_string(),
1968                timestamp_ms: 1515100,
1969            }],
1970            refs: HashMap::new(),
1971            statistics: HashMap::new(),
1972            partition_statistics: HashMap::new(),
1973            encryption_keys: HashMap::from_iter(vec![("key1".to_string(), encryption_key)]),
1974            next_row_id: 5,
1975        };
1976
1977        let expected_json_value = serde_json::to_value(&expected).unwrap();
1978        check_table_metadata_serde(data, expected);
1979
1980        let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
1981        assert_eq!(json_value, expected_json_value);
1982    }
1983
1984    #[test]
1985    fn test_table_data_v1() {
1986        let data = r#"
1987        {
1988            "format-version" : 1,
1989            "table-uuid" : "df838b92-0b32-465d-a44e-d39936e538b7",
1990            "location" : "/home/iceberg/warehouse/nyc/taxis",
1991            "last-updated-ms" : 1662532818843,
1992            "last-column-id" : 5,
1993            "schema" : {
1994              "type" : "struct",
1995              "schema-id" : 0,
1996              "fields" : [ {
1997                "id" : 1,
1998                "name" : "vendor_id",
1999                "required" : false,
2000                "type" : "long"
2001              }, {
2002                "id" : 2,
2003                "name" : "trip_id",
2004                "required" : false,
2005                "type" : "long"
2006              }, {
2007                "id" : 3,
2008                "name" : "trip_distance",
2009                "required" : false,
2010                "type" : "float"
2011              }, {
2012                "id" : 4,
2013                "name" : "fare_amount",
2014                "required" : false,
2015                "type" : "double"
2016              }, {
2017                "id" : 5,
2018                "name" : "store_and_fwd_flag",
2019                "required" : false,
2020                "type" : "string"
2021              } ]
2022            },
2023            "partition-spec" : [ {
2024              "name" : "vendor_id",
2025              "transform" : "identity",
2026              "source-id" : 1,
2027              "field-id" : 1000
2028            } ],
2029            "last-partition-id" : 1000,
2030            "default-sort-order-id" : 0,
2031            "sort-orders" : [ {
2032              "order-id" : 0,
2033              "fields" : [ ]
2034            } ],
2035            "properties" : {
2036              "owner" : "root"
2037            },
2038            "current-snapshot-id" : 638933773299822130,
2039            "refs" : {
2040              "main" : {
2041                "snapshot-id" : 638933773299822130,
2042                "type" : "branch"
2043              }
2044            },
2045            "snapshots" : [ {
2046              "snapshot-id" : 638933773299822130,
2047              "timestamp-ms" : 1662532818843,
2048              "sequence-number" : 0,
2049              "summary" : {
2050                "operation" : "append",
2051                "spark.app.id" : "local-1662532784305",
2052                "added-data-files" : "4",
2053                "added-records" : "4",
2054                "added-files-size" : "6001"
2055              },
2056              "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
2057              "schema-id" : 0
2058            } ],
2059            "snapshot-log" : [ {
2060              "timestamp-ms" : 1662532818843,
2061              "snapshot-id" : 638933773299822130
2062            } ],
2063            "metadata-log" : [ {
2064              "timestamp-ms" : 1662532805245,
2065              "metadata-file" : "/home/iceberg/warehouse/nyc/taxis/metadata/00000-8a62c37d-4573-4021-952a-c0baef7d21d0.metadata.json"
2066            } ]
2067          }
2068        "#;
2069
2070        let schema = Schema::builder()
2071            .with_fields(vec![
2072                Arc::new(NestedField::optional(
2073                    1,
2074                    "vendor_id",
2075                    Type::Primitive(PrimitiveType::Long),
2076                )),
2077                Arc::new(NestedField::optional(
2078                    2,
2079                    "trip_id",
2080                    Type::Primitive(PrimitiveType::Long),
2081                )),
2082                Arc::new(NestedField::optional(
2083                    3,
2084                    "trip_distance",
2085                    Type::Primitive(PrimitiveType::Float),
2086                )),
2087                Arc::new(NestedField::optional(
2088                    4,
2089                    "fare_amount",
2090                    Type::Primitive(PrimitiveType::Double),
2091                )),
2092                Arc::new(NestedField::optional(
2093                    5,
2094                    "store_and_fwd_flag",
2095                    Type::Primitive(PrimitiveType::String),
2096                )),
2097            ])
2098            .build()
2099            .unwrap();
2100
2101        let schema = Arc::new(schema);
2102        let partition_spec = PartitionSpec::builder(schema.clone())
2103            .with_spec_id(0)
2104            .add_partition_field("vendor_id", "vendor_id", Transform::Identity)
2105            .unwrap()
2106            .build()
2107            .unwrap();
2108
2109        let sort_order = SortOrder::builder()
2110            .with_order_id(0)
2111            .build_unbound()
2112            .unwrap();
2113
2114        let snapshot = Snapshot::builder()
2115            .with_snapshot_id(638933773299822130)
2116            .with_timestamp_ms(1662532818843)
2117            .with_sequence_number(0)
2118            .with_schema_id(0)
2119            .with_manifest_list("/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro")
2120            .with_summary(Summary { operation: Operation::Append, additional_properties: HashMap::from_iter(vec![("spark.app.id".to_string(), "local-1662532784305".to_string()), ("added-data-files".to_string(), "4".to_string()), ("added-records".to_string(), "4".to_string()), ("added-files-size".to_string(), "6001".to_string())]) })
2121            .build();
2122
2123        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
2124        let expected = TableMetadata {
2125            format_version: FormatVersion::V1,
2126            table_uuid: Uuid::parse_str("df838b92-0b32-465d-a44e-d39936e538b7").unwrap(),
2127            location: "/home/iceberg/warehouse/nyc/taxis".to_string(),
2128            last_updated_ms: 1662532818843,
2129            last_column_id: 5,
2130            schemas: HashMap::from_iter(vec![(0, schema)]),
2131            current_schema_id: 0,
2132            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
2133            default_partition_type,
2134            default_spec: Arc::new(partition_spec),
2135            last_partition_id: 1000,
2136            default_sort_order_id: 0,
2137            sort_orders: HashMap::from_iter(vec![(0, sort_order.into())]),
2138            snapshots: HashMap::from_iter(vec![(638933773299822130, Arc::new(snapshot))]),
2139            current_snapshot_id: Some(638933773299822130),
2140            last_sequence_number: 0,
2141            properties: HashMap::from_iter(vec![("owner".to_string(), "root".to_string())]),
2142            snapshot_log: vec![SnapshotLog {
2143                snapshot_id: 638933773299822130,
2144                timestamp_ms: 1662532818843,
2145            }],
2146            metadata_log: vec![MetadataLog { metadata_file: "/home/iceberg/warehouse/nyc/taxis/metadata/00000-8a62c37d-4573-4021-952a-c0baef7d21d0.metadata.json".to_string(), timestamp_ms: 1662532805245 }],
2147            refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference { snapshot_id: 638933773299822130, retention: SnapshotRetention::Branch { min_snapshots_to_keep: None, max_snapshot_age_ms: None, max_ref_age_ms: None } })]),
2148            statistics: HashMap::new(),
2149            partition_statistics: HashMap::new(),
2150            encryption_keys: HashMap::new(),
2151            next_row_id: INITIAL_ROW_ID,
2152        };
2153
2154        check_table_metadata_serde(data, expected);
2155    }
2156
2157    #[test]
2158    fn test_table_data_v2_no_snapshots() {
2159        let data = r#"
2160        {
2161            "format-version" : 2,
2162            "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
2163            "location": "s3://b/wh/data.db/table",
2164            "last-sequence-number" : 1,
2165            "last-updated-ms": 1515100955770,
2166            "last-column-id": 1,
2167            "schemas": [
2168                {
2169                    "schema-id" : 1,
2170                    "type" : "struct",
2171                    "fields" :[
2172                        {
2173                            "id": 1,
2174                            "name": "struct_name",
2175                            "required": true,
2176                            "type": "fixed[1]"
2177                        }
2178                    ]
2179                }
2180            ],
2181            "current-schema-id" : 1,
2182            "partition-specs": [
2183                {
2184                    "spec-id": 0,
2185                    "fields": []
2186                }
2187            ],
2188            "refs": {},
2189            "default-spec-id": 0,
2190            "last-partition-id": 1000,
2191            "metadata-log": [
2192                {
2193                    "metadata-file": "s3://bucket/.../v1.json",
2194                    "timestamp-ms": 1515100
2195                }
2196            ],
2197            "sort-orders": [
2198                {
2199                "order-id": 0,
2200                "fields": []
2201                }
2202            ],
2203            "default-sort-order-id": 0
2204        }
2205        "#;
2206
2207        let schema = Schema::builder()
2208            .with_schema_id(1)
2209            .with_fields(vec![Arc::new(NestedField::required(
2210                1,
2211                "struct_name",
2212                Type::Primitive(PrimitiveType::Fixed(1)),
2213            ))])
2214            .build()
2215            .unwrap();
2216
2217        let partition_spec = PartitionSpec::builder(schema.clone())
2218            .with_spec_id(0)
2219            .build()
2220            .unwrap();
2221
2222        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
2223        let expected = TableMetadata {
2224            format_version: FormatVersion::V2,
2225            table_uuid: Uuid::parse_str("fb072c92-a02b-11e9-ae9c-1bb7bc9eca94").unwrap(),
2226            location: "s3://b/wh/data.db/table".to_string(),
2227            last_updated_ms: 1515100955770,
2228            last_column_id: 1,
2229            schemas: HashMap::from_iter(vec![(1, Arc::new(schema))]),
2230            current_schema_id: 1,
2231            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
2232            default_partition_type,
2233            default_spec: partition_spec.into(),
2234            last_partition_id: 1000,
2235            default_sort_order_id: 0,
2236            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
2237            snapshots: HashMap::default(),
2238            current_snapshot_id: None,
2239            last_sequence_number: 1,
2240            properties: HashMap::new(),
2241            snapshot_log: Vec::new(),
2242            metadata_log: vec![MetadataLog {
2243                metadata_file: "s3://bucket/.../v1.json".to_string(),
2244                timestamp_ms: 1515100,
2245            }],
2246            refs: HashMap::new(),
2247            statistics: HashMap::new(),
2248            partition_statistics: HashMap::new(),
2249            encryption_keys: HashMap::new(),
2250            next_row_id: INITIAL_ROW_ID,
2251        };
2252
2253        let expected_json_value = serde_json::to_value(&expected).unwrap();
2254        check_table_metadata_serde(data, expected);
2255
2256        let json_value = serde_json::from_str::<serde_json::Value>(data).unwrap();
2257        assert_eq!(json_value, expected_json_value);
2258    }
2259
2260    #[test]
2261    fn test_current_snapshot_id_must_match_main_branch() {
2262        let data = r#"
2263        {
2264            "format-version" : 2,
2265            "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
2266            "location": "s3://b/wh/data.db/table",
2267            "last-sequence-number" : 1,
2268            "last-updated-ms": 1515100955770,
2269            "last-column-id": 1,
2270            "schemas": [
2271                {
2272                    "schema-id" : 1,
2273                    "type" : "struct",
2274                    "fields" :[
2275                        {
2276                            "id": 1,
2277                            "name": "struct_name",
2278                            "required": true,
2279                            "type": "fixed[1]"
2280                        },
2281                        {
2282                            "id": 4,
2283                            "name": "ts",
2284                            "required": true,
2285                            "type": "timestamp"
2286                        }
2287                    ]
2288                }
2289            ],
2290            "current-schema-id" : 1,
2291            "partition-specs": [
2292                {
2293                    "spec-id": 0,
2294                    "fields": [
2295                        {
2296                            "source-id": 4,
2297                            "field-id": 1000,
2298                            "name": "ts_day",
2299                            "transform": "day"
2300                        }
2301                    ]
2302                }
2303            ],
2304            "default-spec-id": 0,
2305            "last-partition-id": 1000,
2306            "properties": {
2307                "commit.retry.num-retries": "1"
2308            },
2309            "metadata-log": [
2310                {
2311                    "metadata-file": "s3://bucket/.../v1.json",
2312                    "timestamp-ms": 1515100
2313                }
2314            ],
2315            "sort-orders": [
2316                {
2317                "order-id": 0,
2318                "fields": []
2319                }
2320            ],
2321            "default-sort-order-id": 0,
2322            "current-snapshot-id" : 1,
2323            "refs" : {
2324              "main" : {
2325                "snapshot-id" : 2,
2326                "type" : "branch"
2327              }
2328            },
2329            "snapshots" : [ {
2330              "snapshot-id" : 1,
2331              "timestamp-ms" : 1662532818843,
2332              "sequence-number" : 0,
2333              "summary" : {
2334                "operation" : "append",
2335                "spark.app.id" : "local-1662532784305",
2336                "added-data-files" : "4",
2337                "added-records" : "4",
2338                "added-files-size" : "6001"
2339              },
2340              "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
2341              "schema-id" : 0
2342            },
2343            {
2344              "snapshot-id" : 2,
2345              "timestamp-ms" : 1662532818844,
2346              "sequence-number" : 0,
2347              "summary" : {
2348                "operation" : "append",
2349                "spark.app.id" : "local-1662532784305",
2350                "added-data-files" : "4",
2351                "added-records" : "4",
2352                "added-files-size" : "6001"
2353              },
2354              "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
2355              "schema-id" : 0
2356            } ]
2357        }
2358    "#;
2359
2360        let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
2361        assert!(
2362            err.to_string()
2363                .contains("Current snapshot id does not match main branch")
2364        );
2365    }
2366
2367    #[test]
2368    fn test_main_without_current() {
2369        let data = r#"
2370        {
2371            "format-version" : 2,
2372            "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
2373            "location": "s3://b/wh/data.db/table",
2374            "last-sequence-number" : 1,
2375            "last-updated-ms": 1515100955770,
2376            "last-column-id": 1,
2377            "schemas": [
2378                {
2379                    "schema-id" : 1,
2380                    "type" : "struct",
2381                    "fields" :[
2382                        {
2383                            "id": 1,
2384                            "name": "struct_name",
2385                            "required": true,
2386                            "type": "fixed[1]"
2387                        },
2388                        {
2389                            "id": 4,
2390                            "name": "ts",
2391                            "required": true,
2392                            "type": "timestamp"
2393                        }
2394                    ]
2395                }
2396            ],
2397            "current-schema-id" : 1,
2398            "partition-specs": [
2399                {
2400                    "spec-id": 0,
2401                    "fields": [
2402                        {
2403                            "source-id": 4,
2404                            "field-id": 1000,
2405                            "name": "ts_day",
2406                            "transform": "day"
2407                        }
2408                    ]
2409                }
2410            ],
2411            "default-spec-id": 0,
2412            "last-partition-id": 1000,
2413            "properties": {
2414                "commit.retry.num-retries": "1"
2415            },
2416            "metadata-log": [
2417                {
2418                    "metadata-file": "s3://bucket/.../v1.json",
2419                    "timestamp-ms": 1515100
2420                }
2421            ],
2422            "sort-orders": [
2423                {
2424                "order-id": 0,
2425                "fields": []
2426                }
2427            ],
2428            "default-sort-order-id": 0,
2429            "refs" : {
2430              "main" : {
2431                "snapshot-id" : 1,
2432                "type" : "branch"
2433              }
2434            },
2435            "snapshots" : [ {
2436              "snapshot-id" : 1,
2437              "timestamp-ms" : 1662532818843,
2438              "sequence-number" : 0,
2439              "summary" : {
2440                "operation" : "append",
2441                "spark.app.id" : "local-1662532784305",
2442                "added-data-files" : "4",
2443                "added-records" : "4",
2444                "added-files-size" : "6001"
2445              },
2446              "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
2447              "schema-id" : 0
2448            } ]
2449        }
2450    "#;
2451
2452        let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
2453        assert!(
2454            err.to_string()
2455                .contains("Current snapshot is not set, but main branch exists")
2456        );
2457    }
2458
2459    #[test]
2460    fn test_branch_snapshot_missing() {
2461        let data = r#"
2462        {
2463            "format-version" : 2,
2464            "table-uuid": "fb072c92-a02b-11e9-ae9c-1bb7bc9eca94",
2465            "location": "s3://b/wh/data.db/table",
2466            "last-sequence-number" : 1,
2467            "last-updated-ms": 1515100955770,
2468            "last-column-id": 1,
2469            "schemas": [
2470                {
2471                    "schema-id" : 1,
2472                    "type" : "struct",
2473                    "fields" :[
2474                        {
2475                            "id": 1,
2476                            "name": "struct_name",
2477                            "required": true,
2478                            "type": "fixed[1]"
2479                        },
2480                        {
2481                            "id": 4,
2482                            "name": "ts",
2483                            "required": true,
2484                            "type": "timestamp"
2485                        }
2486                    ]
2487                }
2488            ],
2489            "current-schema-id" : 1,
2490            "partition-specs": [
2491                {
2492                    "spec-id": 0,
2493                    "fields": [
2494                        {
2495                            "source-id": 4,
2496                            "field-id": 1000,
2497                            "name": "ts_day",
2498                            "transform": "day"
2499                        }
2500                    ]
2501                }
2502            ],
2503            "default-spec-id": 0,
2504            "last-partition-id": 1000,
2505            "properties": {
2506                "commit.retry.num-retries": "1"
2507            },
2508            "metadata-log": [
2509                {
2510                    "metadata-file": "s3://bucket/.../v1.json",
2511                    "timestamp-ms": 1515100
2512                }
2513            ],
2514            "sort-orders": [
2515                {
2516                "order-id": 0,
2517                "fields": []
2518                }
2519            ],
2520            "default-sort-order-id": 0,
2521            "refs" : {
2522              "main" : {
2523                "snapshot-id" : 1,
2524                "type" : "branch"
2525              },
2526              "foo" : {
2527                "snapshot-id" : 2,
2528                "type" : "branch"
2529              }
2530            },
2531            "snapshots" : [ {
2532              "snapshot-id" : 1,
2533              "timestamp-ms" : 1662532818843,
2534              "sequence-number" : 0,
2535              "summary" : {
2536                "operation" : "append",
2537                "spark.app.id" : "local-1662532784305",
2538                "added-data-files" : "4",
2539                "added-records" : "4",
2540                "added-files-size" : "6001"
2541              },
2542              "manifest-list" : "/home/iceberg/warehouse/nyc/taxis/metadata/snap-638933773299822130-1-7e6760f0-4f6c-4b23-b907-0a5a174e3863.avro",
2543              "schema-id" : 0
2544            } ]
2545        }
2546    "#;
2547
2548        let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
2549        assert!(
2550            err.to_string().contains(
2551                "Snapshot for reference foo does not exist in the existing snapshots list"
2552            )
2553        );
2554    }
2555
2556    #[test]
2557    fn test_v2_wrong_max_snapshot_sequence_number() {
2558        let data = r#"
2559        {
2560            "format-version": 2,
2561            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
2562            "location": "s3://bucket/test/location",
2563            "last-sequence-number": 1,
2564            "last-updated-ms": 1602638573590,
2565            "last-column-id": 3,
2566            "current-schema-id": 0,
2567            "schemas": [
2568                {
2569                    "type": "struct",
2570                    "schema-id": 0,
2571                    "fields": [
2572                        {
2573                            "id": 1,
2574                            "name": "x",
2575                            "required": true,
2576                            "type": "long"
2577                        }
2578                    ]
2579                }
2580            ],
2581            "default-spec-id": 0,
2582            "partition-specs": [
2583                {
2584                    "spec-id": 0,
2585                    "fields": []
2586                }
2587            ],
2588            "last-partition-id": 1000,
2589            "default-sort-order-id": 0,
2590            "sort-orders": [
2591                {
2592                    "order-id": 0,
2593                    "fields": []
2594                }
2595            ],
2596            "properties": {},
2597            "current-snapshot-id": 3055729675574597004,
2598            "snapshots": [
2599                {
2600                    "snapshot-id": 3055729675574597004,
2601                    "timestamp-ms": 1555100955770,
2602                    "sequence-number": 4,
2603                    "summary": {
2604                        "operation": "append"
2605                    },
2606                    "manifest-list": "s3://a/b/2.avro",
2607                    "schema-id": 0
2608                }
2609            ],
2610            "statistics": [],
2611            "snapshot-log": [],
2612            "metadata-log": []
2613        }
2614    "#;
2615
2616        let err = serde_json::from_str::<TableMetadata>(data).unwrap_err();
2617        assert!(err.to_string().contains(
2618            "Invalid snapshot with id 3055729675574597004 and sequence number 4 greater than last sequence number 1"
2619        ));
2620
2621        // Change max sequence number to 4 - should work
2622        let data = data.replace(
2623            r#""last-sequence-number": 1,"#,
2624            r#""last-sequence-number": 4,"#,
2625        );
2626        let metadata = serde_json::from_str::<TableMetadata>(data.as_str()).unwrap();
2627        assert_eq!(metadata.last_sequence_number, 4);
2628
2629        // Change max sequence number to 5 - should work
2630        let data = data.replace(
2631            r#""last-sequence-number": 4,"#,
2632            r#""last-sequence-number": 5,"#,
2633        );
2634        let metadata = serde_json::from_str::<TableMetadata>(data.as_str()).unwrap();
2635        assert_eq!(metadata.last_sequence_number, 5);
2636    }
2637
2638    #[test]
2639    fn test_statistic_files() {
2640        let data = r#"
2641        {
2642            "format-version": 2,
2643            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
2644            "location": "s3://bucket/test/location",
2645            "last-sequence-number": 34,
2646            "last-updated-ms": 1602638573590,
2647            "last-column-id": 3,
2648            "current-schema-id": 0,
2649            "schemas": [
2650                {
2651                    "type": "struct",
2652                    "schema-id": 0,
2653                    "fields": [
2654                        {
2655                            "id": 1,
2656                            "name": "x",
2657                            "required": true,
2658                            "type": "long"
2659                        }
2660                    ]
2661                }
2662            ],
2663            "default-spec-id": 0,
2664            "partition-specs": [
2665                {
2666                    "spec-id": 0,
2667                    "fields": []
2668                }
2669            ],
2670            "last-partition-id": 1000,
2671            "default-sort-order-id": 0,
2672            "sort-orders": [
2673                {
2674                    "order-id": 0,
2675                    "fields": []
2676                }
2677            ],
2678            "properties": {},
2679            "current-snapshot-id": 3055729675574597004,
2680            "snapshots": [
2681                {
2682                    "snapshot-id": 3055729675574597004,
2683                    "timestamp-ms": 1555100955770,
2684                    "sequence-number": 1,
2685                    "summary": {
2686                        "operation": "append"
2687                    },
2688                    "manifest-list": "s3://a/b/2.avro",
2689                    "schema-id": 0
2690                }
2691            ],
2692            "statistics": [
2693                {
2694                    "snapshot-id": 3055729675574597004,
2695                    "statistics-path": "s3://a/b/stats.puffin",
2696                    "file-size-in-bytes": 413,
2697                    "file-footer-size-in-bytes": 42,
2698                    "blob-metadata": [
2699                        {
2700                            "type": "ndv",
2701                            "snapshot-id": 3055729675574597004,
2702                            "sequence-number": 1,
2703                            "fields": [
2704                                1
2705                            ]
2706                        }
2707                    ]
2708                }
2709            ],
2710            "snapshot-log": [],
2711            "metadata-log": []
2712        }
2713    "#;
2714
2715        let schema = Schema::builder()
2716            .with_schema_id(0)
2717            .with_fields(vec![Arc::new(NestedField::required(
2718                1,
2719                "x",
2720                Type::Primitive(PrimitiveType::Long),
2721            ))])
2722            .build()
2723            .unwrap();
2724        let partition_spec = PartitionSpec::builder(schema.clone())
2725            .with_spec_id(0)
2726            .build()
2727            .unwrap();
2728        let snapshot = Snapshot::builder()
2729            .with_snapshot_id(3055729675574597004)
2730            .with_timestamp_ms(1555100955770)
2731            .with_sequence_number(1)
2732            .with_manifest_list("s3://a/b/2.avro")
2733            .with_schema_id(0)
2734            .with_summary(Summary {
2735                operation: Operation::Append,
2736                additional_properties: HashMap::new(),
2737            })
2738            .build();
2739
2740        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
2741        let expected = TableMetadata {
2742            format_version: FormatVersion::V2,
2743            table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
2744            location: "s3://bucket/test/location".to_string(),
2745            last_updated_ms: 1602638573590,
2746            last_column_id: 3,
2747            schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
2748            current_schema_id: 0,
2749            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
2750            default_partition_type,
2751            default_spec: Arc::new(partition_spec),
2752            last_partition_id: 1000,
2753            default_sort_order_id: 0,
2754            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
2755            snapshots: HashMap::from_iter(vec![(3055729675574597004, Arc::new(snapshot))]),
2756            current_snapshot_id: Some(3055729675574597004),
2757            last_sequence_number: 34,
2758            properties: HashMap::new(),
2759            snapshot_log: Vec::new(),
2760            metadata_log: Vec::new(),
2761            statistics: HashMap::from_iter(vec![(3055729675574597004, StatisticsFile {
2762                snapshot_id: 3055729675574597004,
2763                statistics_path: "s3://a/b/stats.puffin".to_string(),
2764                file_size_in_bytes: 413,
2765                file_footer_size_in_bytes: 42,
2766                key_metadata: None,
2767                blob_metadata: vec![BlobMetadata {
2768                    snapshot_id: 3055729675574597004,
2769                    sequence_number: 1,
2770                    fields: vec![1],
2771                    r#type: "ndv".to_string(),
2772                    properties: HashMap::new(),
2773                }],
2774            })]),
2775            partition_statistics: HashMap::new(),
2776            refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
2777                snapshot_id: 3055729675574597004,
2778                retention: SnapshotRetention::Branch {
2779                    min_snapshots_to_keep: None,
2780                    max_snapshot_age_ms: None,
2781                    max_ref_age_ms: None,
2782                },
2783            })]),
2784            encryption_keys: HashMap::new(),
2785            next_row_id: INITIAL_ROW_ID,
2786        };
2787
2788        check_table_metadata_serde(data, expected);
2789    }
2790
2791    #[test]
2792    fn test_partition_statistics_file() {
2793        let data = r#"
2794        {
2795            "format-version": 2,
2796            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
2797            "location": "s3://bucket/test/location",
2798            "last-sequence-number": 34,
2799            "last-updated-ms": 1602638573590,
2800            "last-column-id": 3,
2801            "current-schema-id": 0,
2802            "schemas": [
2803                {
2804                    "type": "struct",
2805                    "schema-id": 0,
2806                    "fields": [
2807                        {
2808                            "id": 1,
2809                            "name": "x",
2810                            "required": true,
2811                            "type": "long"
2812                        }
2813                    ]
2814                }
2815            ],
2816            "default-spec-id": 0,
2817            "partition-specs": [
2818                {
2819                    "spec-id": 0,
2820                    "fields": []
2821                }
2822            ],
2823            "last-partition-id": 1000,
2824            "default-sort-order-id": 0,
2825            "sort-orders": [
2826                {
2827                    "order-id": 0,
2828                    "fields": []
2829                }
2830            ],
2831            "properties": {},
2832            "current-snapshot-id": 3055729675574597004,
2833            "snapshots": [
2834                {
2835                    "snapshot-id": 3055729675574597004,
2836                    "timestamp-ms": 1555100955770,
2837                    "sequence-number": 1,
2838                    "summary": {
2839                        "operation": "append"
2840                    },
2841                    "manifest-list": "s3://a/b/2.avro",
2842                    "schema-id": 0
2843                }
2844            ],
2845            "partition-statistics": [
2846                {
2847                    "snapshot-id": 3055729675574597004,
2848                    "statistics-path": "s3://a/b/partition-stats.parquet",
2849                    "file-size-in-bytes": 43
2850                }
2851            ],
2852            "snapshot-log": [],
2853            "metadata-log": []
2854        }
2855        "#;
2856
2857        let schema = Schema::builder()
2858            .with_schema_id(0)
2859            .with_fields(vec![Arc::new(NestedField::required(
2860                1,
2861                "x",
2862                Type::Primitive(PrimitiveType::Long),
2863            ))])
2864            .build()
2865            .unwrap();
2866        let partition_spec = PartitionSpec::builder(schema.clone())
2867            .with_spec_id(0)
2868            .build()
2869            .unwrap();
2870        let snapshot = Snapshot::builder()
2871            .with_snapshot_id(3055729675574597004)
2872            .with_timestamp_ms(1555100955770)
2873            .with_sequence_number(1)
2874            .with_manifest_list("s3://a/b/2.avro")
2875            .with_schema_id(0)
2876            .with_summary(Summary {
2877                operation: Operation::Append,
2878                additional_properties: HashMap::new(),
2879            })
2880            .build();
2881
2882        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
2883        let expected = TableMetadata {
2884            format_version: FormatVersion::V2,
2885            table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
2886            location: "s3://bucket/test/location".to_string(),
2887            last_updated_ms: 1602638573590,
2888            last_column_id: 3,
2889            schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
2890            current_schema_id: 0,
2891            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
2892            default_spec: Arc::new(partition_spec),
2893            default_partition_type,
2894            last_partition_id: 1000,
2895            default_sort_order_id: 0,
2896            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
2897            snapshots: HashMap::from_iter(vec![(3055729675574597004, Arc::new(snapshot))]),
2898            current_snapshot_id: Some(3055729675574597004),
2899            last_sequence_number: 34,
2900            properties: HashMap::new(),
2901            snapshot_log: Vec::new(),
2902            metadata_log: Vec::new(),
2903            statistics: HashMap::new(),
2904            partition_statistics: HashMap::from_iter(vec![(
2905                3055729675574597004,
2906                PartitionStatisticsFile {
2907                    snapshot_id: 3055729675574597004,
2908                    statistics_path: "s3://a/b/partition-stats.parquet".to_string(),
2909                    file_size_in_bytes: 43,
2910                },
2911            )]),
2912            refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
2913                snapshot_id: 3055729675574597004,
2914                retention: SnapshotRetention::Branch {
2915                    min_snapshots_to_keep: None,
2916                    max_snapshot_age_ms: None,
2917                    max_ref_age_ms: None,
2918                },
2919            })]),
2920            encryption_keys: HashMap::new(),
2921            next_row_id: INITIAL_ROW_ID,
2922        };
2923
2924        check_table_metadata_serde(data, expected);
2925    }
2926
2927    #[test]
2928    fn test_invalid_table_uuid() -> Result<()> {
2929        let data = r#"
2930            {
2931                "format-version" : 2,
2932                "table-uuid": "xxxx"
2933            }
2934        "#;
2935        assert!(serde_json::from_str::<TableMetadata>(data).is_err());
2936        Ok(())
2937    }
2938
2939    #[test]
2940    fn test_deserialize_table_data_v2_invalid_format_version() -> Result<()> {
2941        let data = r#"
2942            {
2943                "format-version" : 1
2944            }
2945        "#;
2946        assert!(serde_json::from_str::<TableMetadata>(data).is_err());
2947        Ok(())
2948    }
2949
2950    #[test]
2951    fn test_table_metadata_v3_valid_minimal() {
2952        let metadata_str =
2953            fs::read_to_string("testdata/table_metadata/TableMetadataV3ValidMinimal.json").unwrap();
2954
2955        let table_metadata = serde_json::from_str::<TableMetadata>(&metadata_str).unwrap();
2956        assert_eq!(table_metadata.format_version, FormatVersion::V3);
2957
2958        let schema = Schema::builder()
2959            .with_schema_id(0)
2960            .with_fields(vec![
2961                Arc::new(
2962                    NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long))
2963                        .with_initial_default(Literal::Primitive(PrimitiveLiteral::Long(1)))
2964                        .with_write_default(Literal::Primitive(PrimitiveLiteral::Long(1))),
2965                ),
2966                Arc::new(
2967                    NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
2968                        .with_doc("comment"),
2969                ),
2970                Arc::new(NestedField::required(
2971                    3,
2972                    "z",
2973                    Type::Primitive(PrimitiveType::Long),
2974                )),
2975            ])
2976            .build()
2977            .unwrap();
2978
2979        let partition_spec = PartitionSpec::builder(schema.clone())
2980            .with_spec_id(0)
2981            .add_unbound_field(UnboundPartitionField {
2982                name: "x".to_string(),
2983                transform: Transform::Identity,
2984                source_id: 1,
2985                field_id: Some(1000),
2986            })
2987            .unwrap()
2988            .build()
2989            .unwrap();
2990
2991        let sort_order = SortOrder::builder()
2992            .with_order_id(3)
2993            .with_sort_field(SortField {
2994                source_id: 2,
2995                transform: Transform::Identity,
2996                direction: SortDirection::Ascending,
2997                null_order: NullOrder::First,
2998            })
2999            .with_sort_field(SortField {
3000                source_id: 3,
3001                transform: Transform::Bucket(4),
3002                direction: SortDirection::Descending,
3003                null_order: NullOrder::Last,
3004            })
3005            .build_unbound()
3006            .unwrap();
3007
3008        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
3009        let expected = TableMetadata {
3010            format_version: FormatVersion::V3,
3011            table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
3012            location: "s3://bucket/test/location".to_string(),
3013            last_updated_ms: 1602638573590,
3014            last_column_id: 3,
3015            schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
3016            current_schema_id: 0,
3017            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
3018            default_spec: Arc::new(partition_spec),
3019            default_partition_type,
3020            last_partition_id: 1000,
3021            default_sort_order_id: 3,
3022            sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
3023            snapshots: HashMap::default(),
3024            current_snapshot_id: None,
3025            last_sequence_number: 34,
3026            properties: HashMap::new(),
3027            snapshot_log: Vec::new(),
3028            metadata_log: Vec::new(),
3029            refs: HashMap::new(),
3030            statistics: HashMap::new(),
3031            partition_statistics: HashMap::new(),
3032            encryption_keys: HashMap::new(),
3033            next_row_id: 0, // V3 specific field from the JSON
3034        };
3035
3036        check_table_metadata_serde(&metadata_str, expected);
3037    }
3038
3039    #[test]
3040    fn test_table_metadata_v2_file_valid() {
3041        let metadata =
3042            fs::read_to_string("testdata/table_metadata/TableMetadataV2Valid.json").unwrap();
3043
3044        let schema1 = Schema::builder()
3045            .with_schema_id(0)
3046            .with_fields(vec![Arc::new(NestedField::required(
3047                1,
3048                "x",
3049                Type::Primitive(PrimitiveType::Long),
3050            ))])
3051            .build()
3052            .unwrap();
3053
3054        let schema2 = Schema::builder()
3055            .with_schema_id(1)
3056            .with_fields(vec![
3057                Arc::new(NestedField::required(
3058                    1,
3059                    "x",
3060                    Type::Primitive(PrimitiveType::Long),
3061                )),
3062                Arc::new(
3063                    NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
3064                        .with_doc("comment"),
3065                ),
3066                Arc::new(NestedField::required(
3067                    3,
3068                    "z",
3069                    Type::Primitive(PrimitiveType::Long),
3070                )),
3071            ])
3072            .with_identifier_field_ids(vec![1, 2])
3073            .build()
3074            .unwrap();
3075
3076        let partition_spec = PartitionSpec::builder(schema2.clone())
3077            .with_spec_id(0)
3078            .add_unbound_field(UnboundPartitionField {
3079                name: "x".to_string(),
3080                transform: Transform::Identity,
3081                source_id: 1,
3082                field_id: Some(1000),
3083            })
3084            .unwrap()
3085            .build()
3086            .unwrap();
3087
3088        let sort_order = SortOrder::builder()
3089            .with_order_id(3)
3090            .with_sort_field(SortField {
3091                source_id: 2,
3092                transform: Transform::Identity,
3093                direction: SortDirection::Ascending,
3094                null_order: NullOrder::First,
3095            })
3096            .with_sort_field(SortField {
3097                source_id: 3,
3098                transform: Transform::Bucket(4),
3099                direction: SortDirection::Descending,
3100                null_order: NullOrder::Last,
3101            })
3102            .build_unbound()
3103            .unwrap();
3104
3105        let snapshot1 = Snapshot::builder()
3106            .with_snapshot_id(3051729675574597004)
3107            .with_timestamp_ms(1515100955770)
3108            .with_sequence_number(0)
3109            .with_manifest_list("s3://a/b/1.avro")
3110            .with_summary(Summary {
3111                operation: Operation::Append,
3112                additional_properties: HashMap::new(),
3113            })
3114            .build();
3115
3116        let snapshot2 = Snapshot::builder()
3117            .with_snapshot_id(3055729675574597004)
3118            .with_parent_snapshot_id(Some(3051729675574597004))
3119            .with_timestamp_ms(1555100955770)
3120            .with_sequence_number(1)
3121            .with_schema_id(1)
3122            .with_manifest_list("s3://a/b/2.avro")
3123            .with_summary(Summary {
3124                operation: Operation::Append,
3125                additional_properties: HashMap::new(),
3126            })
3127            .build();
3128
3129        let default_partition_type = partition_spec.partition_type(&schema2).unwrap();
3130        let expected = TableMetadata {
3131            format_version: FormatVersion::V2,
3132            table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
3133            location: "s3://bucket/test/location".to_string(),
3134            last_updated_ms: 1602638573590,
3135            last_column_id: 3,
3136            schemas: HashMap::from_iter(vec![(0, Arc::new(schema1)), (1, Arc::new(schema2))]),
3137            current_schema_id: 1,
3138            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
3139            default_spec: Arc::new(partition_spec),
3140            default_partition_type,
3141            last_partition_id: 1000,
3142            default_sort_order_id: 3,
3143            sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
3144            snapshots: HashMap::from_iter(vec![
3145                (3051729675574597004, Arc::new(snapshot1)),
3146                (3055729675574597004, Arc::new(snapshot2)),
3147            ]),
3148            current_snapshot_id: Some(3055729675574597004),
3149            last_sequence_number: 34,
3150            properties: HashMap::new(),
3151            snapshot_log: vec![
3152                SnapshotLog {
3153                    snapshot_id: 3051729675574597004,
3154                    timestamp_ms: 1515100955770,
3155                },
3156                SnapshotLog {
3157                    snapshot_id: 3055729675574597004,
3158                    timestamp_ms: 1555100955770,
3159                },
3160            ],
3161            metadata_log: Vec::new(),
3162            refs: HashMap::from_iter(vec![("main".to_string(), SnapshotReference {
3163                snapshot_id: 3055729675574597004,
3164                retention: SnapshotRetention::Branch {
3165                    min_snapshots_to_keep: None,
3166                    max_snapshot_age_ms: None,
3167                    max_ref_age_ms: None,
3168                },
3169            })]),
3170            statistics: HashMap::new(),
3171            partition_statistics: HashMap::new(),
3172            encryption_keys: HashMap::new(),
3173            next_row_id: INITIAL_ROW_ID,
3174        };
3175
3176        check_table_metadata_serde(&metadata, expected);
3177    }
3178
3179    #[test]
3180    fn test_table_metadata_v2_file_valid_minimal() {
3181        let metadata =
3182            fs::read_to_string("testdata/table_metadata/TableMetadataV2ValidMinimal.json").unwrap();
3183
3184        let schema = Schema::builder()
3185            .with_schema_id(0)
3186            .with_fields(vec![
3187                Arc::new(NestedField::required(
3188                    1,
3189                    "x",
3190                    Type::Primitive(PrimitiveType::Long),
3191                )),
3192                Arc::new(
3193                    NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
3194                        .with_doc("comment"),
3195                ),
3196                Arc::new(NestedField::required(
3197                    3,
3198                    "z",
3199                    Type::Primitive(PrimitiveType::Long),
3200                )),
3201            ])
3202            .build()
3203            .unwrap();
3204
3205        let partition_spec = PartitionSpec::builder(schema.clone())
3206            .with_spec_id(0)
3207            .add_unbound_field(UnboundPartitionField {
3208                name: "x".to_string(),
3209                transform: Transform::Identity,
3210                source_id: 1,
3211                field_id: Some(1000),
3212            })
3213            .unwrap()
3214            .build()
3215            .unwrap();
3216
3217        let sort_order = SortOrder::builder()
3218            .with_order_id(3)
3219            .with_sort_field(SortField {
3220                source_id: 2,
3221                transform: Transform::Identity,
3222                direction: SortDirection::Ascending,
3223                null_order: NullOrder::First,
3224            })
3225            .with_sort_field(SortField {
3226                source_id: 3,
3227                transform: Transform::Bucket(4),
3228                direction: SortDirection::Descending,
3229                null_order: NullOrder::Last,
3230            })
3231            .build_unbound()
3232            .unwrap();
3233
3234        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
3235        let expected = TableMetadata {
3236            format_version: FormatVersion::V2,
3237            table_uuid: Uuid::parse_str("9c12d441-03fe-4693-9a96-a0705ddf69c1").unwrap(),
3238            location: "s3://bucket/test/location".to_string(),
3239            last_updated_ms: 1602638573590,
3240            last_column_id: 3,
3241            schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
3242            current_schema_id: 0,
3243            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
3244            default_partition_type,
3245            default_spec: Arc::new(partition_spec),
3246            last_partition_id: 1000,
3247            default_sort_order_id: 3,
3248            sort_orders: HashMap::from_iter(vec![(3, sort_order.into())]),
3249            snapshots: HashMap::default(),
3250            current_snapshot_id: None,
3251            last_sequence_number: 34,
3252            properties: HashMap::new(),
3253            snapshot_log: vec![],
3254            metadata_log: Vec::new(),
3255            refs: HashMap::new(),
3256            statistics: HashMap::new(),
3257            partition_statistics: HashMap::new(),
3258            encryption_keys: HashMap::new(),
3259            next_row_id: INITIAL_ROW_ID,
3260        };
3261
3262        check_table_metadata_serde(&metadata, expected);
3263    }
3264
3265    #[test]
3266    fn test_table_metadata_v1_file_valid() {
3267        let metadata =
3268            fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap();
3269
3270        let schema = Schema::builder()
3271            .with_schema_id(0)
3272            .with_fields(vec![
3273                Arc::new(NestedField::required(
3274                    1,
3275                    "x",
3276                    Type::Primitive(PrimitiveType::Long),
3277                )),
3278                Arc::new(
3279                    NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long))
3280                        .with_doc("comment"),
3281                ),
3282                Arc::new(NestedField::required(
3283                    3,
3284                    "z",
3285                    Type::Primitive(PrimitiveType::Long),
3286                )),
3287            ])
3288            .build()
3289            .unwrap();
3290
3291        let partition_spec = PartitionSpec::builder(schema.clone())
3292            .with_spec_id(0)
3293            .add_unbound_field(UnboundPartitionField {
3294                name: "x".to_string(),
3295                transform: Transform::Identity,
3296                source_id: 1,
3297                field_id: Some(1000),
3298            })
3299            .unwrap()
3300            .build()
3301            .unwrap();
3302
3303        let default_partition_type = partition_spec.partition_type(&schema).unwrap();
3304        let expected = TableMetadata {
3305            format_version: FormatVersion::V1,
3306            table_uuid: Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap(),
3307            location: "s3://bucket/test/location".to_string(),
3308            last_updated_ms: 1602638573874,
3309            last_column_id: 3,
3310            schemas: HashMap::from_iter(vec![(0, Arc::new(schema))]),
3311            current_schema_id: 0,
3312            partition_specs: HashMap::from_iter(vec![(0, partition_spec.clone().into())]),
3313            default_spec: Arc::new(partition_spec),
3314            default_partition_type,
3315            last_partition_id: 0,
3316            default_sort_order_id: 0,
3317            // Sort order is added during deserialization for V2 compatibility
3318            sort_orders: HashMap::from_iter(vec![(0, SortOrder::unsorted_order().into())]),
3319            snapshots: HashMap::new(),
3320            current_snapshot_id: None,
3321            last_sequence_number: 0,
3322            properties: HashMap::new(),
3323            snapshot_log: vec![],
3324            metadata_log: Vec::new(),
3325            refs: HashMap::new(),
3326            statistics: HashMap::new(),
3327            partition_statistics: HashMap::new(),
3328            encryption_keys: HashMap::new(),
3329            next_row_id: INITIAL_ROW_ID,
3330        };
3331
3332        check_table_metadata_serde(&metadata, expected);
3333    }
3334
3335    #[test]
3336    fn test_empty_snapshot_id_is_normalized_to_none() {
3337        let metadata =
3338            fs::read_to_string("testdata/table_metadata/TableMetadataV1Valid.json").unwrap();
3339        let deserialized: TableMetadata = serde_json::from_str(&metadata).unwrap();
3340        assert_eq!(
3341            deserialized.current_snapshot_id(),
3342            None,
3343            "current_snapshot_id of -1 should be deserialized as None"
3344        );
3345    }
3346
3347    #[test]
3348    fn test_table_metadata_v1_compat() {
3349        let metadata =
3350            fs::read_to_string("testdata/table_metadata/TableMetadataV1Compat.json").unwrap();
3351
3352        // Deserialize the JSON to verify it works
3353        let desered_type: TableMetadata = serde_json::from_str(&metadata)
3354            .expect("Failed to deserialize TableMetadataV1Compat.json");
3355
3356        // Verify some key fields match
3357        assert_eq!(desered_type.format_version(), FormatVersion::V1);
3358        assert_eq!(
3359            desered_type.uuid(),
3360            Uuid::parse_str("3276010d-7b1d-488c-98d8-9025fc4fde6b").unwrap()
3361        );
3362        assert_eq!(
3363            desered_type.location(),
3364            "s3://bucket/warehouse/iceberg/glue.db/table_name"
3365        );
3366        assert_eq!(desered_type.last_updated_ms(), 1727773114005);
3367        assert_eq!(desered_type.current_schema_id(), 0);
3368    }
3369
3370    #[test]
3371    fn test_table_metadata_v1_schemas_without_current_id() {
3372        let metadata = fs::read_to_string(
3373            "testdata/table_metadata/TableMetadataV1SchemasWithoutCurrentId.json",
3374        )
3375        .unwrap();
3376
3377        // Deserialize the JSON - this should succeed by using the 'schema' field instead of 'schemas'
3378        let desered_type: TableMetadata = serde_json::from_str(&metadata)
3379            .expect("Failed to deserialize TableMetadataV1SchemasWithoutCurrentId.json");
3380
3381        // Verify it used the 'schema' field
3382        assert_eq!(desered_type.format_version(), FormatVersion::V1);
3383        assert_eq!(
3384            desered_type.uuid(),
3385            Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap()
3386        );
3387
3388        // Get the schema and verify it has the expected fields
3389        let schema = desered_type.current_schema();
3390        assert_eq!(schema.as_struct().fields().len(), 3);
3391        assert_eq!(schema.as_struct().fields()[0].name, "x");
3392        assert_eq!(schema.as_struct().fields()[1].name, "y");
3393        assert_eq!(schema.as_struct().fields()[2].name, "z");
3394    }
3395
3396    #[test]
3397    fn test_table_metadata_v1_no_valid_schema() {
3398        let metadata =
3399            fs::read_to_string("testdata/table_metadata/TableMetadataV1NoValidSchema.json")
3400                .unwrap();
3401
3402        // Deserialize the JSON - this should fail because neither schemas + current_schema_id nor schema is valid
3403        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3404
3405        assert!(desered.is_err());
3406        let error_message = desered.unwrap_err().to_string();
3407        assert!(
3408            error_message.contains("No valid schema configuration found"),
3409            "Expected error about no valid schema configuration, got: {error_message}"
3410        );
3411    }
3412
3413    #[test]
3414    fn test_table_metadata_v1_partition_specs_without_default_id() {
3415        let metadata = fs::read_to_string(
3416            "testdata/table_metadata/TableMetadataV1PartitionSpecsWithoutDefaultId.json",
3417        )
3418        .unwrap();
3419
3420        // Deserialize the JSON - this should succeed by inferring default_spec_id as the max spec ID
3421        let desered_type: TableMetadata = serde_json::from_str(&metadata)
3422            .expect("Failed to deserialize TableMetadataV1PartitionSpecsWithoutDefaultId.json");
3423
3424        // Verify basic metadata
3425        assert_eq!(desered_type.format_version(), FormatVersion::V1);
3426        assert_eq!(
3427            desered_type.uuid(),
3428            Uuid::parse_str("d20125c8-7284-442c-9aea-15fee620737c").unwrap()
3429        );
3430
3431        // Verify partition specs
3432        assert_eq!(desered_type.default_partition_spec_id(), 2); // Should pick the largest spec ID (2)
3433        assert_eq!(desered_type.partition_specs.len(), 2);
3434
3435        // Verify the default spec has the expected fields
3436        let default_spec = &desered_type.default_spec;
3437        assert_eq!(default_spec.spec_id(), 2);
3438        assert_eq!(default_spec.fields().len(), 1);
3439        assert_eq!(default_spec.fields()[0].name, "y");
3440        assert_eq!(default_spec.fields()[0].transform, Transform::Identity);
3441        assert_eq!(default_spec.fields()[0].source_id, 2);
3442    }
3443
3444    #[test]
3445    fn test_table_metadata_v2_schema_not_found() {
3446        let metadata =
3447            fs::read_to_string("testdata/table_metadata/TableMetadataV2CurrentSchemaNotFound.json")
3448                .unwrap();
3449
3450        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3451
3452        assert_eq!(
3453            desered.unwrap_err().to_string(),
3454            "DataInvalid => No schema exists with the current schema id 2."
3455        )
3456    }
3457
3458    #[test]
3459    fn test_table_metadata_v2_missing_sort_order() {
3460        let metadata =
3461            fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingSortOrder.json")
3462                .unwrap();
3463
3464        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3465
3466        assert_eq!(
3467            desered.unwrap_err().to_string(),
3468            "data did not match any variant of untagged enum TableMetadataEnum"
3469        )
3470    }
3471
3472    #[test]
3473    fn test_table_metadata_v2_missing_partition_specs() {
3474        let metadata =
3475            fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingPartitionSpecs.json")
3476                .unwrap();
3477
3478        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3479
3480        assert_eq!(
3481            desered.unwrap_err().to_string(),
3482            "data did not match any variant of untagged enum TableMetadataEnum"
3483        )
3484    }
3485
3486    #[test]
3487    fn test_table_metadata_v2_missing_last_partition_id() {
3488        let metadata = fs::read_to_string(
3489            "testdata/table_metadata/TableMetadataV2MissingLastPartitionId.json",
3490        )
3491        .unwrap();
3492
3493        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3494
3495        assert_eq!(
3496            desered.unwrap_err().to_string(),
3497            "data did not match any variant of untagged enum TableMetadataEnum"
3498        )
3499    }
3500
3501    #[test]
3502    fn test_table_metadata_v2_missing_schemas() {
3503        let metadata =
3504            fs::read_to_string("testdata/table_metadata/TableMetadataV2MissingSchemas.json")
3505                .unwrap();
3506
3507        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3508
3509        assert_eq!(
3510            desered.unwrap_err().to_string(),
3511            "data did not match any variant of untagged enum TableMetadataEnum"
3512        )
3513    }
3514
3515    #[test]
3516    fn test_table_metadata_v2_unsupported_version() {
3517        let metadata =
3518            fs::read_to_string("testdata/table_metadata/TableMetadataUnsupportedVersion.json")
3519                .unwrap();
3520
3521        let desered: Result<TableMetadata, serde_json::Error> = serde_json::from_str(&metadata);
3522
3523        assert_eq!(
3524            desered.unwrap_err().to_string(),
3525            "data did not match any variant of untagged enum TableMetadataEnum"
3526        )
3527    }
3528
3529    #[test]
3530    fn test_order_of_format_version() {
3531        assert!(FormatVersion::V1 < FormatVersion::V2);
3532        assert_eq!(FormatVersion::V1, FormatVersion::V1);
3533        assert_eq!(FormatVersion::V2, FormatVersion::V2);
3534    }
3535
3536    #[test]
3537    fn test_default_partition_spec() {
3538        let default_spec_id = 1234;
3539        let mut table_meta_data = get_test_table_metadata("TableMetadataV2Valid.json");
3540        let partition_spec = PartitionSpec::unpartition_spec();
3541        table_meta_data.default_spec = partition_spec.clone().into();
3542        table_meta_data
3543            .partition_specs
3544            .insert(default_spec_id, Arc::new(partition_spec));
3545
3546        assert_eq!(
3547            (*table_meta_data.default_partition_spec().clone()).clone(),
3548            (*table_meta_data
3549                .partition_spec_by_id(default_spec_id)
3550                .unwrap()
3551                .clone())
3552            .clone()
3553        );
3554    }
3555    #[test]
3556    fn test_default_sort_order() {
3557        let default_sort_order_id = 1234;
3558        let mut table_meta_data = get_test_table_metadata("TableMetadataV2Valid.json");
3559        table_meta_data.default_sort_order_id = default_sort_order_id;
3560        table_meta_data
3561            .sort_orders
3562            .insert(default_sort_order_id, Arc::new(SortOrder::default()));
3563
3564        assert_eq!(
3565            table_meta_data.default_sort_order(),
3566            table_meta_data
3567                .sort_orders
3568                .get(&default_sort_order_id)
3569                .unwrap()
3570        )
3571    }
3572
3573    #[test]
3574    fn test_table_metadata_builder_from_table_creation() {
3575        let table_creation = TableCreation::builder()
3576            .location("s3://db/table".to_string())
3577            .name("table".to_string())
3578            .properties(HashMap::new())
3579            .schema(Schema::builder().build().unwrap())
3580            .build();
3581        let table_metadata = TableMetadataBuilder::from_table_creation(table_creation)
3582            .unwrap()
3583            .build()
3584            .unwrap()
3585            .metadata;
3586        assert_eq!(table_metadata.location, "s3://db/table");
3587        assert_eq!(table_metadata.schemas.len(), 1);
3588        assert_eq!(
3589            table_metadata
3590                .schemas
3591                .get(&0)
3592                .unwrap()
3593                .as_struct()
3594                .fields()
3595                .len(),
3596            0
3597        );
3598        assert_eq!(table_metadata.properties.len(), 0);
3599        assert_eq!(
3600            table_metadata.partition_specs,
3601            HashMap::from([(
3602                0,
3603                Arc::new(
3604                    PartitionSpec::builder(table_metadata.schemas.get(&0).unwrap().clone())
3605                        .with_spec_id(0)
3606                        .build()
3607                        .unwrap()
3608                )
3609            )])
3610        );
3611        assert_eq!(
3612            table_metadata.sort_orders,
3613            HashMap::from([(
3614                0,
3615                Arc::new(SortOrder {
3616                    order_id: 0,
3617                    fields: vec![]
3618                })
3619            )])
3620        );
3621    }
3622
3623    #[tokio::test]
3624    async fn test_table_metadata_read_write() {
3625        // Create a temporary directory for our test
3626        let temp_dir = TempDir::new().unwrap();
3627        let temp_path = temp_dir.path().to_str().unwrap();
3628
3629        // Create a FileIO instance
3630        let file_io = FileIO::new_with_fs();
3631
3632        // Use an existing test metadata from the test files
3633        let original_metadata: TableMetadata =
3634            get_test_table_metadata_at("TableMetadataV2Valid.json", temp_path);
3635
3636        // Define the metadata location
3637        let metadata_location =
3638            MetadataLocation::try_new_with_metadata(&original_metadata).unwrap();
3639        let metadata_location_str = metadata_location.to_string();
3640
3641        // Write the metadata
3642        original_metadata
3643            .write_to(&file_io, &metadata_location)
3644            .await
3645            .unwrap();
3646
3647        // Verify the file exists
3648        assert!(fs::metadata(&metadata_location_str).is_ok());
3649
3650        // Read the metadata back
3651        let read_metadata = TableMetadata::read_from(&file_io, &metadata_location_str)
3652            .await
3653            .unwrap();
3654
3655        // Verify the metadata matches
3656        assert_eq!(read_metadata, original_metadata);
3657    }
3658
3659    #[tokio::test]
3660    async fn test_table_metadata_read_compressed() {
3661        let temp_dir = TempDir::new().unwrap();
3662        let metadata_location = temp_dir.path().join("v1.gz.metadata.json");
3663
3664        let original_metadata: TableMetadata = get_test_table_metadata("TableMetadataV2Valid.json");
3665        let json = serde_json::to_string(&original_metadata).unwrap();
3666
3667        let compressed = CompressionCodec::gzip_default()
3668            .compress(json.into_bytes())
3669            .expect("failed to compress metadata");
3670        fs::write(&metadata_location, &compressed).expect("failed to write metadata");
3671
3672        // Read the metadata back
3673        let file_io = FileIO::new_with_fs();
3674        let metadata_location = metadata_location.to_str().unwrap();
3675        let read_metadata = TableMetadata::read_from(&file_io, metadata_location)
3676            .await
3677            .unwrap();
3678
3679        // Verify the metadata matches
3680        assert_eq!(read_metadata, original_metadata);
3681    }
3682
3683    #[tokio::test]
3684    async fn test_table_metadata_read_nonexistent_file() {
3685        // Create a FileIO instance
3686        let file_io = FileIO::new_with_fs();
3687
3688        // Try to read a non-existent file
3689        let result = TableMetadata::read_from(&file_io, "/nonexistent/path/metadata.json").await;
3690
3691        // Verify it returns an error
3692        assert!(result.is_err());
3693    }
3694
3695    #[tokio::test]
3696    async fn test_table_metadata_write_with_gzip_compression() {
3697        let temp_dir = TempDir::new().unwrap();
3698        let temp_path = temp_dir.path().to_str().unwrap();
3699        let file_io = FileIO::new_with_fs();
3700
3701        // Get a test metadata and add gzip compression property
3702        let original_metadata: TableMetadata =
3703            get_test_table_metadata_at("TableMetadataV2Valid.json", temp_path);
3704
3705        // Modify properties to enable gzip compression (using mixed case to test case-insensitive matching)
3706        let mut props = original_metadata.properties.clone();
3707        props.insert(
3708            TableProperties::PROPERTY_METADATA_COMPRESSION_CODEC.to_string(),
3709            "GziP".to_string(),
3710        );
3711        // Use builder to create new metadata with updated properties
3712        let compressed_metadata =
3713            TableMetadataBuilder::new_from_metadata(original_metadata.clone(), None)
3714                .assign_uuid(original_metadata.table_uuid)
3715                .set_properties(props.clone())
3716                .unwrap()
3717                .build()
3718                .unwrap()
3719                .metadata;
3720
3721        // Create MetadataLocation with compression codec from metadata
3722        let metadata_location =
3723            MetadataLocation::try_new_with_metadata(&compressed_metadata).unwrap();
3724        let metadata_location_str = metadata_location.to_string();
3725
3726        // Verify the location has the .gz extension
3727        assert!(metadata_location_str.contains(".gz.metadata.json"));
3728
3729        // Write the metadata with compression
3730        compressed_metadata
3731            .write_to(&file_io, &metadata_location)
3732            .await
3733            .unwrap();
3734
3735        // Verify the compressed file exists
3736        assert!(std::path::Path::new(&metadata_location_str).exists());
3737
3738        // Read the raw file and check it's gzip compressed
3739        let raw_content = fs::read(&metadata_location_str).unwrap();
3740        assert!(raw_content.len() > 2);
3741        assert_eq!(raw_content[0], 0x1F); // gzip magic number
3742        assert_eq!(raw_content[1], 0x8B); // gzip magic number
3743
3744        // Read the metadata back using the compressed location
3745        let read_metadata = TableMetadata::read_from(&file_io, &metadata_location_str)
3746            .await
3747            .unwrap();
3748
3749        // Verify the complete round-trip: read metadata should match what we wrote
3750        assert_eq!(read_metadata, compressed_metadata);
3751    }
3752
3753    #[test]
3754    fn test_partition_name_exists() {
3755        let schema = Schema::builder()
3756            .with_fields(vec![
3757                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3758                NestedField::required(2, "partition_col", Type::Primitive(PrimitiveType::Int))
3759                    .into(),
3760            ])
3761            .build()
3762            .unwrap();
3763
3764        let spec1 = PartitionSpec::builder(schema.clone())
3765            .with_spec_id(1)
3766            .add_partition_field("data", "data_partition", Transform::Identity)
3767            .unwrap()
3768            .build()
3769            .unwrap();
3770
3771        let spec2 = PartitionSpec::builder(schema.clone())
3772            .with_spec_id(2)
3773            .add_partition_field("partition_col", "partition_bucket", Transform::Bucket(16))
3774            .unwrap()
3775            .build()
3776            .unwrap();
3777
3778        // Build metadata with these specs
3779        let metadata = TableMetadataBuilder::new(
3780            schema,
3781            spec1.clone().into_unbound(),
3782            SortOrder::unsorted_order(),
3783            "s3://test/location".to_string(),
3784            FormatVersion::V2,
3785            HashMap::new(),
3786        )
3787        .unwrap()
3788        .add_partition_spec(spec2.into_unbound())
3789        .unwrap()
3790        .build()
3791        .unwrap()
3792        .metadata;
3793
3794        assert!(metadata.partition_name_exists("data_partition"));
3795        assert!(metadata.partition_name_exists("partition_bucket"));
3796
3797        assert!(!metadata.partition_name_exists("nonexistent_field"));
3798        assert!(!metadata.partition_name_exists("data")); // schema field name, not partition field name
3799        assert!(!metadata.partition_name_exists(""));
3800    }
3801
3802    #[test]
3803    fn test_partition_name_exists_empty_specs() {
3804        // Create metadata with no partition specs (unpartitioned table)
3805        let schema = Schema::builder()
3806            .with_fields(vec![
3807                NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3808            ])
3809            .build()
3810            .unwrap();
3811
3812        let metadata = TableMetadataBuilder::new(
3813            schema,
3814            PartitionSpec::unpartition_spec().into_unbound(),
3815            SortOrder::unsorted_order(),
3816            "s3://test/location".to_string(),
3817            FormatVersion::V2,
3818            HashMap::new(),
3819        )
3820        .unwrap()
3821        .build()
3822        .unwrap()
3823        .metadata;
3824
3825        assert!(!metadata.partition_name_exists("any_field"));
3826        assert!(!metadata.partition_name_exists("data"));
3827    }
3828
3829    #[test]
3830    fn test_name_exists_in_any_schema() {
3831        // Create multiple schemas with different fields
3832        let schema1 = Schema::builder()
3833            .with_schema_id(1)
3834            .with_fields(vec![
3835                NestedField::required(1, "field1", Type::Primitive(PrimitiveType::String)).into(),
3836                NestedField::required(2, "field2", Type::Primitive(PrimitiveType::Int)).into(),
3837            ])
3838            .build()
3839            .unwrap();
3840
3841        let schema2 = Schema::builder()
3842            .with_schema_id(2)
3843            .with_fields(vec![
3844                NestedField::required(1, "field1", Type::Primitive(PrimitiveType::String)).into(),
3845                NestedField::required(3, "field3", Type::Primitive(PrimitiveType::Long)).into(),
3846            ])
3847            .build()
3848            .unwrap();
3849
3850        let metadata = TableMetadataBuilder::new(
3851            schema1,
3852            PartitionSpec::unpartition_spec().into_unbound(),
3853            SortOrder::unsorted_order(),
3854            "s3://test/location".to_string(),
3855            FormatVersion::V2,
3856            HashMap::new(),
3857        )
3858        .unwrap()
3859        .add_current_schema(schema2)
3860        .unwrap()
3861        .build()
3862        .unwrap()
3863        .metadata;
3864
3865        assert!(metadata.name_exists_in_any_schema("field1")); // exists in both schemas
3866        assert!(metadata.name_exists_in_any_schema("field2")); // exists only in schema1 (historical)
3867        assert!(metadata.name_exists_in_any_schema("field3")); // exists only in schema2 (current)
3868
3869        assert!(!metadata.name_exists_in_any_schema("nonexistent_field"));
3870        assert!(!metadata.name_exists_in_any_schema("field4"));
3871        assert!(!metadata.name_exists_in_any_schema(""));
3872    }
3873
3874    #[test]
3875    fn test_name_exists_in_any_schema_empty_schemas() {
3876        let schema = Schema::builder().with_fields(vec![]).build().unwrap();
3877
3878        let metadata = TableMetadataBuilder::new(
3879            schema,
3880            PartitionSpec::unpartition_spec().into_unbound(),
3881            SortOrder::unsorted_order(),
3882            "s3://test/location".to_string(),
3883            FormatVersion::V2,
3884            HashMap::new(),
3885        )
3886        .unwrap()
3887        .build()
3888        .unwrap()
3889        .metadata;
3890
3891        assert!(!metadata.name_exists_in_any_schema("any_field"));
3892    }
3893
3894    #[test]
3895    fn test_helper_methods_multi_version_scenario() {
3896        // Test a realistic multi-version scenario
3897        let initial_schema = Schema::builder()
3898            .with_fields(vec![
3899                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
3900                NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
3901                NestedField::required(
3902                    3,
3903                    "deprecated_field",
3904                    Type::Primitive(PrimitiveType::String),
3905                )
3906                .into(),
3907            ])
3908            .build()
3909            .unwrap();
3910
3911        let metadata = TableMetadataBuilder::new(
3912            initial_schema,
3913            PartitionSpec::unpartition_spec().into_unbound(),
3914            SortOrder::unsorted_order(),
3915            "s3://test/location".to_string(),
3916            FormatVersion::V2,
3917            HashMap::new(),
3918        )
3919        .unwrap();
3920
3921        let evolved_schema = Schema::builder()
3922            .with_fields(vec![
3923                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
3924                NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
3925                NestedField::required(
3926                    3,
3927                    "deprecated_field",
3928                    Type::Primitive(PrimitiveType::String),
3929                )
3930                .into(),
3931                NestedField::required(4, "new_field", Type::Primitive(PrimitiveType::Double))
3932                    .into(),
3933            ])
3934            .build()
3935            .unwrap();
3936
3937        // Then add a third schema that removes the deprecated field
3938        let _final_schema = Schema::builder()
3939            .with_fields(vec![
3940                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
3941                NestedField::required(2, "name", Type::Primitive(PrimitiveType::String)).into(),
3942                NestedField::required(4, "new_field", Type::Primitive(PrimitiveType::Double))
3943                    .into(),
3944                NestedField::required(5, "latest_field", Type::Primitive(PrimitiveType::Boolean))
3945                    .into(),
3946            ])
3947            .build()
3948            .unwrap();
3949
3950        let final_metadata = metadata
3951            .add_current_schema(evolved_schema)
3952            .unwrap()
3953            .build()
3954            .unwrap()
3955            .metadata;
3956
3957        assert!(!final_metadata.partition_name_exists("nonexistent_partition")); // unpartitioned table
3958
3959        assert!(final_metadata.name_exists_in_any_schema("id")); // exists in both schemas
3960        assert!(final_metadata.name_exists_in_any_schema("name")); // exists in both schemas
3961        assert!(final_metadata.name_exists_in_any_schema("deprecated_field")); // exists in both schemas
3962        assert!(final_metadata.name_exists_in_any_schema("new_field")); // only in current schema
3963        assert!(!final_metadata.name_exists_in_any_schema("never_existed"));
3964    }
3965
3966    #[test]
3967    fn test_invalid_sort_order_id_zero_with_fields() {
3968        let metadata = r#"
3969        {
3970            "format-version": 2,
3971            "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
3972            "location": "s3://bucket/test/location",
3973            "last-sequence-number": 111,
3974            "last-updated-ms": 1600000000000,
3975            "last-column-id": 3,
3976            "current-schema-id": 1,
3977            "schemas": [
3978                {
3979                    "type": "struct",
3980                    "schema-id": 1,
3981                    "fields": [
3982                        {"id": 1, "name": "x", "required": true, "type": "long"},
3983                        {"id": 2, "name": "y", "required": true, "type": "long"}
3984                    ]
3985                }
3986            ],
3987            "default-spec-id": 0,
3988            "partition-specs": [{"spec-id": 0, "fields": []}],
3989            "last-partition-id": 999,
3990            "default-sort-order-id": 0,
3991            "sort-orders": [
3992                {
3993                    "order-id": 0,
3994                    "fields": [
3995                        {
3996                            "transform": "identity",
3997                            "source-id": 1,
3998                            "direction": "asc",
3999                            "null-order": "nulls-first"
4000                        }
4001                    ]
4002                }
4003            ],
4004            "properties": {},
4005            "current-snapshot-id": -1,
4006            "snapshots": []
4007        }
4008        "#;
4009
4010        let result: Result<TableMetadata, serde_json::Error> = serde_json::from_str(metadata);
4011
4012        // Should fail because sort order ID 0 is reserved for unsorted order and cannot have fields
4013        assert!(
4014            result.is_err(),
4015            "Parsing should fail for sort order ID 0 with fields"
4016        );
4017    }
4018
4019    #[test]
4020    fn test_table_properties_with_defaults() {
4021        use crate::spec::TableProperties;
4022
4023        let schema = Schema::builder()
4024            .with_fields(vec![
4025                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
4026            ])
4027            .build()
4028            .unwrap();
4029
4030        let metadata = TableMetadataBuilder::new(
4031            schema,
4032            PartitionSpec::unpartition_spec().into_unbound(),
4033            SortOrder::unsorted_order(),
4034            "s3://test/location".to_string(),
4035            FormatVersion::V2,
4036            HashMap::new(),
4037        )
4038        .unwrap()
4039        .build()
4040        .unwrap()
4041        .metadata;
4042
4043        let props = metadata.table_properties().unwrap();
4044
4045        assert_eq!(
4046            props.commit_num_retries,
4047            TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
4048        );
4049        assert_eq!(
4050            props.write_target_file_size_bytes,
4051            TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT
4052        );
4053    }
4054
4055    #[test]
4056    fn test_table_properties_with_custom_values() {
4057        use crate::spec::TableProperties;
4058
4059        let schema = Schema::builder()
4060            .with_fields(vec![
4061                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
4062            ])
4063            .build()
4064            .unwrap();
4065
4066        let properties = HashMap::from([
4067            (
4068                TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string(),
4069                "10".to_string(),
4070            ),
4071            (
4072                TableProperties::PROPERTY_WRITE_TARGET_FILE_SIZE_BYTES.to_string(),
4073                "1024".to_string(),
4074            ),
4075        ]);
4076
4077        let metadata = TableMetadataBuilder::new(
4078            schema,
4079            PartitionSpec::unpartition_spec().into_unbound(),
4080            SortOrder::unsorted_order(),
4081            "s3://test/location".to_string(),
4082            FormatVersion::V2,
4083            properties,
4084        )
4085        .unwrap()
4086        .build()
4087        .unwrap()
4088        .metadata;
4089
4090        let props = metadata.table_properties().unwrap();
4091
4092        assert_eq!(props.commit_num_retries, 10);
4093        assert_eq!(props.write_target_file_size_bytes, 1024);
4094    }
4095
4096    #[test]
4097    fn test_table_properties_with_invalid_value() {
4098        let schema = Schema::builder()
4099            .with_fields(vec![
4100                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
4101            ])
4102            .build()
4103            .unwrap();
4104
4105        let properties = HashMap::from([(
4106            "commit.retry.num-retries".to_string(),
4107            "not_a_number".to_string(),
4108        )]);
4109
4110        let metadata = TableMetadataBuilder::new(
4111            schema,
4112            PartitionSpec::unpartition_spec().into_unbound(),
4113            SortOrder::unsorted_order(),
4114            "s3://test/location".to_string(),
4115            FormatVersion::V2,
4116            properties,
4117        )
4118        .unwrap()
4119        .build()
4120        .unwrap()
4121        .metadata;
4122
4123        let err = metadata.table_properties().unwrap_err();
4124        assert_eq!(err.kind(), ErrorKind::DataInvalid);
4125        assert!(err.message().contains("Invalid table properties"));
4126    }
4127
4128    #[test]
4129    fn test_v2_to_v3_upgrade_preserves_existing_snapshots_without_row_lineage() {
4130        // Create a v2 table metadata
4131        let schema = Schema::builder()
4132            .with_fields(vec![
4133                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
4134            ])
4135            .build()
4136            .unwrap();
4137
4138        let v2_metadata = TableMetadataBuilder::new(
4139            schema,
4140            PartitionSpec::unpartition_spec().into_unbound(),
4141            SortOrder::unsorted_order(),
4142            "s3://bucket/test/location".to_string(),
4143            FormatVersion::V2,
4144            HashMap::new(),
4145        )
4146        .unwrap()
4147        .build()
4148        .unwrap()
4149        .metadata;
4150
4151        // Add a v2 snapshot
4152        let snapshot = Snapshot::builder()
4153            .with_snapshot_id(1)
4154            .with_timestamp_ms(v2_metadata.last_updated_ms + 1)
4155            .with_sequence_number(1)
4156            .with_schema_id(0)
4157            .with_manifest_list("s3://bucket/test/metadata/snap-1.avro")
4158            .with_summary(Summary {
4159                operation: Operation::Append,
4160                additional_properties: HashMap::from([(
4161                    "added-data-files".to_string(),
4162                    "1".to_string(),
4163                )]),
4164            })
4165            .build();
4166
4167        let v2_with_snapshot = v2_metadata
4168            .into_builder(Some("s3://bucket/test/metadata/v00001.json".to_string()))
4169            .add_snapshot(snapshot)
4170            .unwrap()
4171            .set_ref("main", SnapshotReference {
4172                snapshot_id: 1,
4173                retention: SnapshotRetention::Branch {
4174                    min_snapshots_to_keep: None,
4175                    max_snapshot_age_ms: None,
4176                    max_ref_age_ms: None,
4177                },
4178            })
4179            .unwrap()
4180            .build()
4181            .unwrap()
4182            .metadata;
4183
4184        // Verify v2 serialization works fine
4185        let v2_json = serde_json::to_string(&v2_with_snapshot);
4186        assert!(v2_json.is_ok(), "v2 serialization should work");
4187
4188        // Upgrade to v3
4189        let v3_metadata = v2_with_snapshot
4190            .into_builder(Some("s3://bucket/test/metadata/v00002.json".to_string()))
4191            .upgrade_format_version(FormatVersion::V3)
4192            .unwrap()
4193            .build()
4194            .unwrap()
4195            .metadata;
4196
4197        assert_eq!(v3_metadata.format_version, FormatVersion::V3);
4198        assert_eq!(v3_metadata.next_row_id, INITIAL_ROW_ID);
4199        assert_eq!(v3_metadata.snapshots.len(), 1);
4200
4201        // Verify the snapshot has no row_range
4202        let snapshot = v3_metadata.snapshots.values().next().unwrap();
4203        assert!(
4204            snapshot.row_range().is_none(),
4205            "Snapshot should have no row_range after upgrade"
4206        );
4207
4208        // Try to serialize v3 metadata - this should now work
4209        let v3_json = serde_json::to_string(&v3_metadata);
4210        assert!(
4211            v3_json.is_ok(),
4212            "v3 serialization should work for upgraded tables"
4213        );
4214
4215        // Verify we can deserialize it back
4216        let deserialized: TableMetadata = serde_json::from_str(&v3_json.unwrap()).unwrap();
4217        assert_eq!(deserialized.format_version, FormatVersion::V3);
4218        assert_eq!(deserialized.snapshots.len(), 1);
4219
4220        // Verify the deserialized snapshot still has no row_range
4221        let deserialized_snapshot = deserialized.snapshots.values().next().unwrap();
4222        assert!(
4223            deserialized_snapshot.row_range().is_none(),
4224            "Deserialized snapshot should have no row_range"
4225        );
4226    }
4227
4228    #[test]
4229    fn test_v3_snapshot_with_row_lineage_serialization() {
4230        // Create a v3 table metadata
4231        let schema = Schema::builder()
4232            .with_fields(vec![
4233                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
4234            ])
4235            .build()
4236            .unwrap();
4237
4238        let v3_metadata = TableMetadataBuilder::new(
4239            schema,
4240            PartitionSpec::unpartition_spec().into_unbound(),
4241            SortOrder::unsorted_order(),
4242            "s3://bucket/test/location".to_string(),
4243            FormatVersion::V3,
4244            HashMap::new(),
4245        )
4246        .unwrap()
4247        .build()
4248        .unwrap()
4249        .metadata;
4250
4251        // Add a v3 snapshot with row lineage
4252        let snapshot = Snapshot::builder()
4253            .with_snapshot_id(1)
4254            .with_timestamp_ms(v3_metadata.last_updated_ms + 1)
4255            .with_sequence_number(1)
4256            .with_schema_id(0)
4257            .with_manifest_list("s3://bucket/test/metadata/snap-1.avro")
4258            .with_summary(Summary {
4259                operation: Operation::Append,
4260                additional_properties: HashMap::from([(
4261                    "added-data-files".to_string(),
4262                    "1".to_string(),
4263                )]),
4264            })
4265            .with_row_range(100, 50) // first_row_id=100, added_rows=50
4266            .build();
4267
4268        let v3_with_snapshot = v3_metadata
4269            .into_builder(Some("s3://bucket/test/metadata/v00001.json".to_string()))
4270            .add_snapshot(snapshot)
4271            .unwrap()
4272            .set_ref("main", SnapshotReference {
4273                snapshot_id: 1,
4274                retention: SnapshotRetention::Branch {
4275                    min_snapshots_to_keep: None,
4276                    max_snapshot_age_ms: None,
4277                    max_ref_age_ms: None,
4278                },
4279            })
4280            .unwrap()
4281            .build()
4282            .unwrap()
4283            .metadata;
4284
4285        // Verify the snapshot has row_range
4286        let snapshot = v3_with_snapshot.snapshots.values().next().unwrap();
4287        assert!(
4288            snapshot.row_range().is_some(),
4289            "Snapshot should have row_range"
4290        );
4291        let (first_row_id, added_rows) = snapshot.row_range().unwrap();
4292        assert_eq!(first_row_id, 100);
4293        assert_eq!(added_rows, 50);
4294
4295        // Serialize v3 metadata - this should work
4296        let v3_json = serde_json::to_string(&v3_with_snapshot);
4297        assert!(
4298            v3_json.is_ok(),
4299            "v3 serialization should work for snapshots with row lineage"
4300        );
4301
4302        // Verify we can deserialize it back
4303        let deserialized: TableMetadata = serde_json::from_str(&v3_json.unwrap()).unwrap();
4304        assert_eq!(deserialized.format_version, FormatVersion::V3);
4305        assert_eq!(deserialized.snapshots.len(), 1);
4306
4307        // Verify the deserialized snapshot has the correct row_range
4308        let deserialized_snapshot = deserialized.snapshots.values().next().unwrap();
4309        assert!(
4310            deserialized_snapshot.row_range().is_some(),
4311            "Deserialized snapshot should have row_range"
4312        );
4313        let (deserialized_first_row_id, deserialized_added_rows) =
4314            deserialized_snapshot.row_range().unwrap();
4315        assert_eq!(deserialized_first_row_id, 100);
4316        assert_eq!(deserialized_added_rows, 50);
4317    }
4318
4319    #[test]
4320    fn test_metadata_location_default() {
4321        // Verify metadata files go to `<location>/metadata` when `write.metadata.path` is not set
4322        let metadata = get_test_table_metadata("TableMetadataV2Valid.json");
4323        assert_eq!(metadata.location(), "s3://bucket/test/location");
4324        assert_eq!(
4325            metadata.metadata_location().unwrap(),
4326            "s3://bucket/test/location/metadata"
4327        );
4328    }
4329
4330    #[test]
4331    fn test_metadata_location_honors_write_metadata_path() {
4332        let metadata = get_test_table_metadata("TableMetadataV2Valid.json")
4333            .into_builder(None)
4334            .set_properties(HashMap::from([(
4335                TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
4336                "s3://other-bucket/custom-meta".to_string(),
4337            )]))
4338            .unwrap()
4339            .build()
4340            .unwrap()
4341            .metadata;
4342        assert_eq!(
4343            metadata.metadata_location().unwrap(),
4344            "s3://other-bucket/custom-meta"
4345        );
4346    }
4347
4348    #[test]
4349    fn test_metadata_location_trims_trailing_slash() {
4350        // A configured path with a trailing slash must not yield a doubled separator
4351        let metadata = get_test_table_metadata("TableMetadataV2Valid.json")
4352            .into_builder(None)
4353            .set_properties(HashMap::from([(
4354                TableProperties::PROPERTY_WRITE_METADATA_PATH.to_string(),
4355                "s3://other-bucket/custom-meta/".to_string(),
4356            )]))
4357            .unwrap()
4358            .build()
4359            .unwrap()
4360            .metadata;
4361        assert_eq!(
4362            metadata.metadata_location().unwrap(),
4363            "s3://other-bucket/custom-meta"
4364        );
4365    }
4366}