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