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