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