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