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