1use std::collections::{HashMap, HashSet};
19use std::sync::Arc;
20
21use uuid::Uuid;
22
23use super::{
24 DEFAULT_PARTITION_SPEC_ID, DEFAULT_SCHEMA_ID, FormatVersion, MAIN_BRANCH, MetadataLog,
25 ONE_MINUTE_MS, PartitionField, PartitionSpec, PartitionSpecBuilder, PartitionStatisticsFile,
26 Schema, SchemaRef, Snapshot, SnapshotLog, SnapshotReference, SnapshotRetention, SortOrder,
27 SortOrderRef, StatisticsFile, StructType, TableMetadata, TableProperties, Transform,
28 UNPARTITIONED_LAST_ASSIGNED_ID, UnboundPartitionSpec,
29};
30use crate::error::{Error, ErrorKind, Result, invalid_data};
31use crate::spec::{EncryptedKey, INITIAL_ROW_ID, MIN_FORMAT_VERSION_ROW_LINEAGE};
32use crate::{TableCreation, TableUpdate};
33
34pub(crate) const FIRST_FIELD_ID: i32 = 1;
35
36#[derive(Debug, Clone)]
49pub struct TableMetadataBuilder {
50 metadata: TableMetadata,
51 changes: Vec<TableUpdate>,
52 last_added_schema_id: Option<i32>,
53 last_added_spec_id: Option<i32>,
54 last_added_order_id: Option<i64>,
55 previous_history_entry: Option<MetadataLog>,
57 last_updated_ms: Option<i64>,
58}
59
60#[derive(Debug, Clone, PartialEq)]
61pub struct TableMetadataBuildResult {
63 pub metadata: TableMetadata,
65 pub changes: Vec<TableUpdate>,
67 pub expired_metadata_logs: Vec<MetadataLog>,
69}
70
71impl TableMetadataBuilder {
72 pub const LAST_ADDED: i32 = -1;
74
75 pub fn new(
80 schema: Schema,
81 spec: impl Into<UnboundPartitionSpec>,
82 sort_order: SortOrder,
83 location: String,
84 format_version: FormatVersion,
85 properties: HashMap<String, String>,
86 ) -> Result<Self> {
87 let (fresh_schema, fresh_spec, fresh_sort_order) =
89 Self::reassign_ids(schema, spec.into(), sort_order)?;
90 let schema_id = fresh_schema.schema_id();
91
92 let builder = Self {
93 metadata: TableMetadata {
94 format_version,
95 table_uuid: Uuid::now_v7(),
96 location: "".to_string(), last_sequence_number: 0,
98 last_updated_ms: 0, last_column_id: -1, current_schema_id: -1, schemas: HashMap::new(),
102 partition_specs: HashMap::new(),
103 default_spec: Arc::new(
104 PartitionSpec::unpartition_spec().with_spec_id(-1),
110 ), default_partition_type: StructType::new(vec![]),
112 last_partition_id: UNPARTITIONED_LAST_ASSIGNED_ID,
113 properties: HashMap::new(),
114 current_snapshot_id: None,
115 snapshots: HashMap::new(),
116 snapshot_log: vec![],
117 sort_orders: HashMap::new(),
118 metadata_log: vec![],
119 default_sort_order_id: -1, refs: HashMap::default(),
121 statistics: HashMap::new(),
122 partition_statistics: HashMap::new(),
123 encryption_keys: HashMap::new(),
124 next_row_id: INITIAL_ROW_ID,
125 },
126 last_updated_ms: None,
127 changes: vec![],
128 last_added_schema_id: Some(schema_id),
129 last_added_spec_id: None,
130 last_added_order_id: None,
131 previous_history_entry: None,
132 };
133
134 builder
135 .set_location(location)
136 .add_current_schema(fresh_schema)?
137 .add_default_partition_spec(fresh_spec.into_unbound())?
138 .add_default_sort_order(fresh_sort_order)?
139 .set_properties(properties)
140 }
141
142 #[must_use]
148 pub fn new_from_metadata(
149 previous: TableMetadata,
150 current_file_location: Option<String>,
151 ) -> Self {
152 Self {
153 previous_history_entry: current_file_location.map(|l| MetadataLog {
154 metadata_file: l,
155 timestamp_ms: previous.last_updated_ms,
156 }),
157 metadata: previous,
158 changes: Vec::default(),
159 last_added_schema_id: None,
160 last_added_spec_id: None,
161 last_added_order_id: None,
162 last_updated_ms: None,
163 }
164 }
165
166 pub fn from_table_creation(table_creation: TableCreation) -> Result<Self> {
168 let TableCreation {
169 name: _,
170 location,
171 schema,
172 partition_spec,
173 sort_order,
174 properties,
175 format_version,
176 } = table_creation;
177
178 let location =
179 location.ok_or_else(|| invalid_data!("Can't create table without location"))?;
180 let partition_spec = partition_spec.unwrap_or(UnboundPartitionSpec {
181 spec_id: None,
182 fields: vec![],
183 });
184
185 Self::new(
186 schema,
187 partition_spec,
188 sort_order.unwrap_or(SortOrder::unsorted_order()),
189 location,
190 format_version,
191 properties,
192 )
193 }
194
195 pub fn assign_uuid(mut self, uuid: Uuid) -> Self {
197 if self.metadata.table_uuid != uuid {
198 self.metadata.table_uuid = uuid;
199 self.changes.push(TableUpdate::AssignUuid { uuid });
200 }
201
202 self
203 }
204
205 pub fn upgrade_format_version(mut self, format_version: FormatVersion) -> Result<Self> {
210 if format_version < self.metadata.format_version {
211 return Err(invalid_data!(
212 "Cannot downgrade FormatVersion from {} to {}",
213 self.metadata.format_version,
214 format_version
215 ));
216 }
217
218 if format_version != self.metadata.format_version {
219 match format_version {
220 FormatVersion::V1 => {
221 }
223 FormatVersion::V2 => {
224 self.metadata.format_version = format_version;
225 self.changes
226 .push(TableUpdate::UpgradeFormatVersion { format_version });
227 }
228 FormatVersion::V3 => {
229 self.metadata.format_version = format_version;
230 self.metadata.next_row_id = INITIAL_ROW_ID;
232 self.changes
233 .push(TableUpdate::UpgradeFormatVersion { format_version });
234 }
235 }
236 }
237
238 Ok(self)
239 }
240
241 pub fn set_properties(mut self, properties: HashMap<String, String>) -> Result<Self> {
250 let reserved_properties = properties
252 .keys()
253 .filter(|key| TableProperties::RESERVED_PROPERTIES.contains(&key.as_str()))
254 .map(ToString::to_string)
255 .collect::<Vec<_>>();
256
257 if !reserved_properties.is_empty() {
258 return Err(invalid_data!(
259 "Table properties should not contain reserved properties, but got: [{}]",
260 reserved_properties.join(", ")
261 ));
262 }
263
264 if properties.is_empty() {
265 return Ok(self);
266 }
267
268 self.metadata.properties.extend(properties.clone());
269 self.changes.push(TableUpdate::SetProperties {
270 updates: properties,
271 });
272
273 Ok(self)
274 }
275
276 pub fn remove_properties(mut self, properties: &[String]) -> Result<Self> {
282 let properties = properties.iter().cloned().collect::<HashSet<_>>();
284
285 let reserved_properties = properties
287 .iter()
288 .filter(|key| TableProperties::RESERVED_PROPERTIES.contains(&key.as_str()))
289 .map(ToString::to_string)
290 .collect::<Vec<_>>();
291
292 if !reserved_properties.is_empty() {
293 return Err(invalid_data!(
294 "Table properties to remove contain reserved properties: [{}]",
295 reserved_properties.join(", ")
296 ));
297 }
298
299 for property in &properties {
300 self.metadata.properties.remove(property);
301 }
302
303 if !properties.is_empty() {
304 self.changes.push(TableUpdate::RemoveProperties {
305 removals: properties.into_iter().collect(),
306 });
307 }
308
309 Ok(self)
310 }
311
312 pub fn set_location(mut self, location: String) -> Self {
314 let location = location.trim_end_matches('/').to_string();
315 if self.metadata.location != location {
316 self.changes.push(TableUpdate::SetLocation {
317 location: location.clone(),
318 });
319 self.metadata.location = location;
320 }
321
322 self
323 }
324
325 pub fn add_snapshot(mut self, snapshot: Snapshot) -> Result<Self> {
334 if self
335 .metadata
336 .snapshots
337 .contains_key(&snapshot.snapshot_id())
338 {
339 return Err(invalid_data!(
340 "Snapshot already exists for: '{}'",
341 snapshot.snapshot_id()
342 ));
343 }
344
345 if self.metadata.format_version != FormatVersion::V1
346 && snapshot.sequence_number() <= self.metadata.last_sequence_number
347 && snapshot.parent_snapshot_id().is_some()
348 {
349 return Err(invalid_data!(
350 "Cannot add snapshot with sequence number {} older than last sequence number {}",
351 snapshot.sequence_number(),
352 self.metadata.last_sequence_number
353 ));
354 }
355
356 if let Some(last) = self.metadata.snapshot_log.last() {
357 if snapshot.timestamp_ms() - last.timestamp_ms < -ONE_MINUTE_MS {
360 return Err(invalid_data!(
361 "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
362 snapshot.timestamp_ms(),
363 last.timestamp_ms
364 ));
365 }
366 }
367
368 let max_last_updated = self
369 .last_updated_ms
370 .unwrap_or_default()
371 .max(self.metadata.last_updated_ms);
372 if snapshot.timestamp_ms() - max_last_updated < -ONE_MINUTE_MS {
373 return Err(invalid_data!(
374 "Invalid snapshot timestamp {}: before last updated timestamp {}",
375 snapshot.timestamp_ms(),
376 max_last_updated
377 ));
378 }
379
380 let mut added_rows = None;
381 if self.metadata.format_version >= MIN_FORMAT_VERSION_ROW_LINEAGE {
382 if let Some((first_row_id, added_rows_count)) = snapshot.row_range() {
383 if first_row_id < self.metadata.next_row_id {
384 return Err(invalid_data!(
385 "Cannot add a snapshot, first-row-id is behind table next-row-id: {first_row_id} < {}",
386 self.metadata.next_row_id
387 ));
388 }
389
390 added_rows = Some(added_rows_count);
391 } else {
392 return Err(invalid_data!(
393 "Cannot add a snapshot: first-row-id is null. first-row-id must be set for format version >= {MIN_FORMAT_VERSION_ROW_LINEAGE}",
394 ));
395 }
396 }
397
398 if let Some(added_rows) = added_rows {
399 self.metadata.next_row_id = self
400 .metadata
401 .next_row_id
402 .checked_add(added_rows)
403 .ok_or_else(|| {
404 invalid_data!(
405 "Cannot add snapshot: next-row-id overflowed when adding added-rows"
406 )
407 })?;
408 }
409
410 self.changes.push(TableUpdate::AddSnapshot {
412 snapshot: snapshot.clone(),
413 });
414
415 self.last_updated_ms = Some(snapshot.timestamp_ms());
416 self.metadata.last_sequence_number = snapshot.sequence_number();
417 self.metadata
418 .snapshots
419 .insert(snapshot.snapshot_id(), snapshot.into());
420
421 Ok(self)
422 }
423
424 pub fn set_branch_snapshot(self, snapshot: Snapshot, branch: &str) -> Result<Self> {
430 let reference = self.metadata.refs.get(branch).cloned();
431
432 let reference = if let Some(mut reference) = reference {
433 if !reference.is_branch() {
434 return Err(invalid_data!(
435 "Cannot append snapshot to non-branch reference '{branch}'",
436 ));
437 }
438
439 reference.snapshot_id = snapshot.snapshot_id();
440 reference
441 } else {
442 SnapshotReference {
443 snapshot_id: snapshot.snapshot_id(),
444 retention: SnapshotRetention::Branch {
445 min_snapshots_to_keep: None,
446 max_snapshot_age_ms: None,
447 max_ref_age_ms: None,
448 },
449 }
450 };
451
452 self.add_snapshot(snapshot)?.set_ref(branch, reference)
453 }
454
455 pub fn remove_snapshots(mut self, snapshot_ids: &[i64]) -> Self {
459 let mut removed_snapshots = Vec::with_capacity(snapshot_ids.len());
460
461 self.metadata.snapshots.retain(|k, _| {
462 if snapshot_ids.contains(k) {
463 removed_snapshots.push(*k);
464 false
465 } else {
466 true
467 }
468 });
469
470 if !removed_snapshots.is_empty() {
471 self.changes.push(TableUpdate::RemoveSnapshots {
472 snapshot_ids: removed_snapshots,
473 });
474 }
475
476 self.metadata
478 .refs
479 .retain(|_, v| self.metadata.snapshots.contains_key(&v.snapshot_id));
480
481 self
482 }
483
484 pub fn set_ref(mut self, ref_name: &str, reference: SnapshotReference) -> Result<Self> {
489 if self
490 .metadata
491 .refs
492 .get(ref_name)
493 .is_some_and(|snap_ref| snap_ref.eq(&reference))
494 {
495 return Ok(self);
496 }
497
498 let Some(snapshot) = self.metadata.snapshots.get(&reference.snapshot_id) else {
499 return Err(invalid_data!(
500 "Cannot set '{ref_name}' to unknown snapshot: '{}'",
501 reference.snapshot_id
502 ));
503 };
504
505 let is_added_snapshot = self.changes.iter().any(|update| {
507 matches!(update, TableUpdate::AddSnapshot { snapshot: snap } if snap.snapshot_id() == snapshot.snapshot_id())
508 });
509 if is_added_snapshot {
510 self.last_updated_ms = Some(snapshot.timestamp_ms());
511 }
512
513 if ref_name == MAIN_BRANCH {
515 self.metadata.current_snapshot_id = Some(snapshot.snapshot_id());
516 let timestamp_ms = if let Some(last_updated_ms) = self.last_updated_ms {
517 last_updated_ms
518 } else {
519 let last_updated_ms = chrono::Utc::now().timestamp_millis();
520 self.last_updated_ms = Some(last_updated_ms);
521 last_updated_ms
522 };
523
524 self.metadata.snapshot_log.push(SnapshotLog {
525 snapshot_id: snapshot.snapshot_id(),
526 timestamp_ms,
527 });
528 }
529
530 self.changes.push(TableUpdate::SetSnapshotRef {
531 ref_name: ref_name.to_string(),
532 reference: reference.clone(),
533 });
534 self.metadata.refs.insert(ref_name.to_string(), reference);
535
536 Ok(self)
537 }
538
539 pub fn remove_ref(mut self, ref_name: &str) -> Self {
543 if ref_name == MAIN_BRANCH {
544 self.metadata.current_snapshot_id = None;
545 }
546
547 if self.metadata.refs.remove(ref_name).is_some() || ref_name == MAIN_BRANCH {
548 self.changes.push(TableUpdate::RemoveSnapshotRef {
549 ref_name: ref_name.to_string(),
550 });
551 }
552
553 self
554 }
555
556 pub fn set_statistics(mut self, statistics: StatisticsFile) -> Self {
558 self.metadata
559 .statistics
560 .insert(statistics.snapshot_id, statistics.clone());
561 self.changes.push(TableUpdate::SetStatistics {
562 statistics: statistics.clone(),
563 });
564 self
565 }
566
567 pub fn remove_statistics(mut self, snapshot_id: i64) -> Self {
569 let previous = self.metadata.statistics.remove(&snapshot_id);
570 if previous.is_some() {
571 self.changes
572 .push(TableUpdate::RemoveStatistics { snapshot_id });
573 }
574 self
575 }
576
577 pub fn set_partition_statistics(
579 mut self,
580 partition_statistics_file: PartitionStatisticsFile,
581 ) -> Self {
582 self.metadata.partition_statistics.insert(
583 partition_statistics_file.snapshot_id,
584 partition_statistics_file.clone(),
585 );
586 self.changes.push(TableUpdate::SetPartitionStatistics {
587 partition_statistics: partition_statistics_file,
588 });
589 self
590 }
591
592 pub fn remove_partition_statistics(mut self, snapshot_id: i64) -> Self {
594 let previous = self.metadata.partition_statistics.remove(&snapshot_id);
595 if previous.is_some() {
596 self.changes
597 .push(TableUpdate::RemovePartitionStatistics { snapshot_id });
598 }
599 self
600 }
601
602 pub fn add_schema(mut self, schema: Schema) -> Result<Self> {
609 let new_schema_id = self.reuse_or_create_new_schema_id(&schema);
610 let schema_found = self.metadata.schemas.contains_key(&new_schema_id);
611
612 if schema_found {
613 if self.last_added_schema_id != Some(new_schema_id) {
614 self.changes.push(TableUpdate::AddSchema {
615 schema: schema.clone(),
616 });
617 self.last_added_schema_id = Some(new_schema_id);
618 }
619
620 return Ok(self);
621 }
622
623 self.metadata.last_column_id =
626 std::cmp::max(self.metadata.last_column_id, schema.highest_field_id());
627
628 let schema = match new_schema_id == schema.schema_id() {
630 true => schema,
631 false => schema.with_schema_id(new_schema_id),
632 };
633
634 self.metadata
635 .schemas
636 .insert(new_schema_id, schema.clone().into());
637
638 self.changes.push(TableUpdate::AddSchema { schema });
639
640 self.last_added_schema_id = Some(new_schema_id);
641
642 Ok(self)
643 }
644
645 pub fn set_current_schema(mut self, mut schema_id: i32) -> Result<Self> {
654 if schema_id == Self::LAST_ADDED {
655 schema_id = self.last_added_schema_id.ok_or_else(|| {
656 invalid_data!(
657 "Cannot set current schema to last added schema: no schema has been added."
658 )
659 })?;
660 };
661 let schema_id = schema_id; if schema_id == self.metadata.current_schema_id {
664 return Ok(self);
665 }
666
667 let schema = self.metadata.schemas.get(&schema_id).ok_or_else(|| {
668 invalid_data!("Cannot set current schema to unknown schema with id: '{schema_id}'")
669 })?;
670 self.validate_schema_field_names(schema)?;
671
672 self.metadata.current_schema_id = schema_id;
678
679 if self.last_added_schema_id == Some(schema_id) {
680 self.changes.push(TableUpdate::SetCurrentSchema {
681 schema_id: Self::LAST_ADDED,
682 });
683 } else {
684 self.changes
685 .push(TableUpdate::SetCurrentSchema { schema_id });
686 }
687
688 Ok(self)
689 }
690
691 pub fn add_current_schema(self, schema: Schema) -> Result<Self> {
693 self.add_schema(schema)?
694 .set_current_schema(Self::LAST_ADDED)
695 }
696
697 fn validate_schema_field_names(&self, schema: &Schema) -> Result<()> {
707 for (spec_id, spec) in &self.metadata.partition_specs {
708 for partition_field in spec.fields() {
709 if schema.field_by_id(partition_field.source_id).is_none() {
710 continue;
711 }
712
713 let Some(schema_field) = schema.field_by_name(&partition_field.name) else {
714 continue;
715 };
716
717 if schema_field.id != partition_field.source_id {
718 return Err(invalid_data!(
719 "Cannot set current schema {}: field '{}' with id {} conflicts with a \
720 partition field of that name in spec {} sourced from field id {}",
721 schema.schema_id(),
722 partition_field.name,
723 schema_field.id,
724 spec_id,
725 partition_field.source_id
726 ));
727 }
728 }
729 }
730
731 Ok(())
732 }
733
734 fn validate_partition_field_names(&self, unbound_spec: &UnboundPartitionSpec) -> Result<()> {
745 if self.metadata.schemas.is_empty() {
746 return Ok(());
747 }
748
749 let current_schema = self.get_current_schema()?;
750 for partition_field in unbound_spec.fields() {
751 let exists_in_any_schema = self
752 .metadata
753 .name_exists_in_any_schema(partition_field.name());
754
755 if !exists_in_any_schema {
757 continue;
758 }
759
760 if let Some(schema_field) = current_schema.field_by_name(partition_field.name()) {
762 let has_matching_source_id = schema_field.id == partition_field.source_id()?;
763
764 if !partition_field.transform().may_share_name_with_source() {
765 return Err(invalid_data!(
766 "Cannot create partition with name '{}' that conflicts with schema field and is not an identity or void transform.",
767 partition_field.name()
768 ));
769 }
770
771 if !has_matching_source_id {
772 return Err(invalid_data!(
773 "Cannot create identity or void partition sourced from different field in schema. \
774 Field name '{}' has id `{}` in schema but partition source id is `{}`",
775 partition_field.name(),
776 schema_field.id,
777 partition_field.source_id()?
778 ));
779 }
780 }
781 }
782
783 Ok(())
784 }
785
786 pub fn add_partition_spec(mut self, unbound_spec: UnboundPartitionSpec) -> Result<Self> {
797 let schema = self.get_current_schema()?.clone();
798
799 self.validate_partition_field_names(&unbound_spec)?;
801
802 let unbound_spec = self.reuse_partition_field_ids(unbound_spec)?;
804
805 let spec = PartitionSpecBuilder::new_from_unbound(unbound_spec.clone(), schema)?
806 .with_last_assigned_field_id(self.metadata.last_partition_id)
807 .build()?;
808
809 let new_spec_id = self.reuse_or_create_new_spec_id(&spec);
810 let spec_found = self.metadata.partition_specs.contains_key(&new_spec_id);
811 let spec = spec.with_spec_id(new_spec_id);
812 let unbound_spec = unbound_spec.with_spec_id(new_spec_id);
813
814 if spec_found {
815 if self.last_added_spec_id != Some(new_spec_id) {
816 self.changes
817 .push(TableUpdate::AddSpec { spec: unbound_spec });
818 self.last_added_spec_id = Some(new_spec_id);
819 }
820
821 return Ok(self);
822 }
823
824 if self.metadata.format_version <= FormatVersion::V1 && !spec.has_sequential_ids() {
825 return Err(invalid_data!(
826 "Cannot add partition spec with non-sequential field ids to format version 1 table"
827 ));
828 }
829
830 let highest_field_id = spec
831 .highest_field_id()
832 .unwrap_or(UNPARTITIONED_LAST_ASSIGNED_ID);
833 self.metadata
834 .partition_specs
835 .insert(new_spec_id, Arc::new(spec));
836 self.changes
837 .push(TableUpdate::AddSpec { spec: unbound_spec });
838
839 self.last_added_spec_id = Some(new_spec_id);
840 self.metadata.last_partition_id =
841 std::cmp::max(self.metadata.last_partition_id, highest_field_id);
842
843 Ok(self)
844 }
845
846 fn reuse_partition_field_ids(
851 &self,
852 unbound_spec: UnboundPartitionSpec,
853 ) -> Result<UnboundPartitionSpec> {
854 let equivalent_field_ids = Self::highest_equivalent_field_ids(
855 self.metadata
856 .partition_specs
857 .values()
858 .flat_map(|spec| spec.fields()),
859 );
860
861 let fields = unbound_spec
863 .fields
864 .into_iter()
865 .map(|field| {
866 if field.field_id().is_none()
867 && let Some(&existing_field_id) =
868 equivalent_field_ids.get(&(field.source_ids().to_vec(), field.transform()))
869 {
870 field.with_field_id(existing_field_id)
871 } else {
872 field
873 }
874 })
875 .collect();
876
877 Ok(UnboundPartitionSpec {
878 spec_id: unbound_spec.spec_id,
879 fields,
880 })
881 }
882
883 fn highest_equivalent_field_ids<'a>(
888 fields: impl Iterator<Item = &'a PartitionField>,
889 ) -> HashMap<(Vec<i32>, Transform), i32> {
890 let mut equivalent_field_ids = HashMap::new();
891 for field in fields {
892 equivalent_field_ids
893 .entry((vec![field.source_id], field.transform))
894 .and_modify(|id: &mut i32| *id = (*id).max(field.field_id))
895 .or_insert(field.field_id);
896 }
897 equivalent_field_ids
898 }
899
900 pub fn set_default_partition_spec(mut self, mut spec_id: i32) -> Result<Self> {
906 if spec_id == Self::LAST_ADDED {
907 spec_id = self.last_added_spec_id.ok_or_else(|| {
908 invalid_data!(
909 "Cannot set default partition spec to last added spec: no spec has been added."
910 )
911 })?;
912 }
913
914 if self.metadata.default_spec.spec_id() == spec_id {
915 return Ok(self);
916 }
917
918 if !self.metadata.partition_specs.contains_key(&spec_id) {
919 return Err(invalid_data!(
920 "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
921 ));
922 }
923
924 let schemaless_spec = self
925 .metadata
926 .partition_specs
927 .get(&spec_id)
928 .ok_or_else(|| {
929 invalid_data!(
930 "Cannot set default partition spec to unknown spec with id: '{spec_id}'",
931 )
932 })?
933 .clone();
934 let spec = Arc::unwrap_or_clone(schemaless_spec);
935 let spec_type = spec.partition_type(self.get_current_schema()?)?;
936 self.metadata.default_spec = Arc::new(spec);
937 self.metadata.default_partition_type = spec_type;
938
939 if self.last_added_spec_id == Some(spec_id) {
940 self.changes.push(TableUpdate::SetDefaultSpec {
941 spec_id: Self::LAST_ADDED,
942 });
943 } else {
944 self.changes.push(TableUpdate::SetDefaultSpec { spec_id });
945 }
946
947 Ok(self)
948 }
949
950 pub fn add_default_partition_spec(self, unbound_spec: UnboundPartitionSpec) -> Result<Self> {
952 self.add_partition_spec(unbound_spec)?
953 .set_default_partition_spec(Self::LAST_ADDED)
954 }
955
956 pub fn remove_partition_specs(mut self, spec_ids: &[i32]) -> Result<Self> {
963 if spec_ids.contains(&self.metadata.default_spec.spec_id()) {
964 return Err(invalid_data!("Cannot remove default partition spec"));
965 }
966
967 let mut removed_specs = Vec::with_capacity(spec_ids.len());
968 spec_ids.iter().for_each(|id| {
969 if self.metadata.partition_specs.remove(id).is_some() {
970 removed_specs.push(*id);
971 }
972 });
973
974 if !removed_specs.is_empty() {
975 self.changes.push(TableUpdate::RemovePartitionSpecs {
976 spec_ids: removed_specs,
977 });
978 }
979
980 Ok(self)
981 }
982
983 pub fn add_sort_order(mut self, sort_order: SortOrder) -> Result<Self> {
994 let new_order_id = self.reuse_or_create_new_sort_id(&sort_order);
995 let sort_order_found = self.metadata.sort_orders.contains_key(&new_order_id);
996
997 if sort_order_found {
998 if self.last_added_order_id != Some(new_order_id) {
999 self.changes.push(TableUpdate::AddSortOrder {
1000 sort_order: sort_order.clone().with_order_id(new_order_id),
1001 });
1002 self.last_added_order_id = Some(new_order_id);
1003 }
1004
1005 return Ok(self);
1006 }
1007
1008 let schema = self.get_current_schema()?.clone().as_ref().clone();
1009 let sort_order = SortOrder::builder()
1010 .with_order_id(new_order_id)
1011 .with_fields(sort_order.fields)
1012 .build(&schema)
1013 .map_err(|e| {
1014 invalid_data!("Sort order to add is incompatible with current schema: {e}")
1015 .with_source(e)
1016 })?;
1017
1018 self.last_added_order_id = Some(new_order_id);
1019 self.metadata
1020 .sort_orders
1021 .insert(new_order_id, sort_order.clone().into());
1022 self.changes.push(TableUpdate::AddSortOrder { sort_order });
1023
1024 Ok(self)
1025 }
1026
1027 pub fn set_default_sort_order(mut self, mut sort_order_id: i64) -> Result<Self> {
1033 if sort_order_id == Self::LAST_ADDED as i64 {
1034 sort_order_id = self.last_added_order_id.ok_or_else(|| {
1035 invalid_data!(
1036 "Cannot set default sort order to last added order: no order has been added."
1037 )
1038 })?;
1039 }
1040
1041 if self.metadata.default_sort_order_id == sort_order_id {
1042 return Ok(self);
1043 }
1044
1045 if !self.metadata.sort_orders.contains_key(&sort_order_id) {
1046 return Err(invalid_data!(
1047 "Cannot set default sort order to unknown order with id: '{sort_order_id}'"
1048 ));
1049 }
1050
1051 self.metadata.default_sort_order_id = sort_order_id;
1052
1053 if self.last_added_order_id == Some(sort_order_id) {
1054 self.changes.push(TableUpdate::SetDefaultSortOrder {
1055 sort_order_id: Self::LAST_ADDED as i64,
1056 });
1057 } else {
1058 self.changes
1059 .push(TableUpdate::SetDefaultSortOrder { sort_order_id });
1060 }
1061
1062 Ok(self)
1063 }
1064
1065 fn add_default_sort_order(self, sort_order: SortOrder) -> Result<Self> {
1067 self.add_sort_order(sort_order)?
1068 .set_default_sort_order(Self::LAST_ADDED as i64)
1069 }
1070
1071 pub fn add_encryption_key(mut self, key: EncryptedKey) -> Self {
1073 let key_id = key.key_id().to_string();
1074 if self.metadata.encryption_keys.contains_key(&key_id) {
1075 return self;
1077 }
1078
1079 self.metadata.encryption_keys.insert(key_id, key.clone());
1080 self.changes.push(TableUpdate::AddEncryptionKey {
1081 encryption_key: key,
1082 });
1083 self
1084 }
1085
1086 pub fn remove_encryption_key(mut self, key_id: &str) -> Self {
1088 if self.metadata.encryption_keys.remove(key_id).is_some() {
1089 self.changes.push(TableUpdate::RemoveEncryptionKey {
1090 key_id: key_id.to_string(),
1091 });
1092 }
1093 self
1094 }
1095
1096 pub fn build(mut self) -> Result<TableMetadataBuildResult> {
1098 self.metadata.last_updated_ms = self
1099 .last_updated_ms
1100 .unwrap_or_else(|| chrono::Utc::now().timestamp_millis());
1101
1102 let schema = self.get_current_schema()?.clone();
1106 let sort_order = Arc::unwrap_or_clone(self.get_default_sort_order()?);
1107
1108 self.metadata.default_spec = Arc::new(
1109 Arc::unwrap_or_clone(self.metadata.default_spec)
1110 .into_unbound()
1111 .bind(schema.clone())?,
1112 );
1113 self.metadata.default_partition_type =
1114 self.metadata.default_spec.partition_type(&schema)?;
1115 SortOrder::builder()
1116 .with_fields(sort_order.fields)
1117 .build(&schema)?;
1118
1119 self.update_snapshot_log()?;
1120 self.metadata.try_normalize()?;
1121
1122 if let Some(hist_entry) = self.previous_history_entry.take() {
1123 self.metadata.metadata_log.push(hist_entry);
1124 }
1125 let expired_metadata_logs = self.expire_metadata_log();
1126
1127 Ok(TableMetadataBuildResult {
1128 metadata: self.metadata,
1129 changes: self.changes,
1130 expired_metadata_logs,
1131 })
1132 }
1133
1134 fn expire_metadata_log(&mut self) -> Vec<MetadataLog> {
1135 let max_size = self
1136 .metadata
1137 .properties
1138 .get(TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX)
1139 .and_then(|v| v.parse::<usize>().ok())
1140 .unwrap_or(TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX_DEFAULT)
1141 .max(1);
1142
1143 if self.metadata.metadata_log.len() > max_size {
1144 self.metadata
1145 .metadata_log
1146 .drain(0..self.metadata.metadata_log.len() - max_size)
1147 .collect()
1148 } else {
1149 Vec::new()
1150 }
1151 }
1152
1153 fn update_snapshot_log(&mut self) -> Result<()> {
1154 let intermediate_snapshots = self.get_intermediate_snapshots();
1155 let has_removed_snapshots = self
1156 .changes
1157 .iter()
1158 .any(|update| matches!(update, TableUpdate::RemoveSnapshots { .. }));
1159
1160 if intermediate_snapshots.is_empty() && !has_removed_snapshots {
1161 return Ok(());
1162 }
1163
1164 let mut new_snapshot_log = Vec::new();
1165 for log_entry in &self.metadata.snapshot_log {
1166 let snapshot_id = log_entry.snapshot_id;
1167 if self.metadata.snapshots.contains_key(&snapshot_id) {
1168 if !intermediate_snapshots.contains(&snapshot_id) {
1169 new_snapshot_log.push(log_entry.clone());
1170 }
1171 } else if has_removed_snapshots {
1172 new_snapshot_log.clear();
1178 }
1179 }
1180
1181 if let Some(current_snapshot_id) = self.metadata.current_snapshot_id {
1182 let last_id = new_snapshot_log.last().map(|entry| entry.snapshot_id);
1183 if last_id != Some(current_snapshot_id) {
1184 return Err(invalid_data!(
1185 "Cannot set invalid snapshot log: latest entry is not the current snapshot"
1186 ));
1187 }
1188 };
1189
1190 self.metadata.snapshot_log = new_snapshot_log;
1191 Ok(())
1192 }
1193
1194 fn get_intermediate_snapshots(&self) -> HashSet<i64> {
1204 let added_snapshot_ids = self
1205 .changes
1206 .iter()
1207 .filter_map(|update| match update {
1208 TableUpdate::AddSnapshot { snapshot } => Some(snapshot.snapshot_id()),
1209 _ => None,
1210 })
1211 .collect::<HashSet<_>>();
1212
1213 self.changes
1214 .iter()
1215 .filter_map(|update| match update {
1216 TableUpdate::SetSnapshotRef {
1217 ref_name,
1218 reference,
1219 } => {
1220 if added_snapshot_ids.contains(&reference.snapshot_id)
1221 && ref_name == MAIN_BRANCH
1222 && reference.snapshot_id
1223 != self
1224 .metadata
1225 .current_snapshot_id
1226 .unwrap_or(i64::from(Self::LAST_ADDED))
1227 {
1228 Some(reference.snapshot_id)
1229 } else {
1230 None
1231 }
1232 }
1233 _ => None,
1234 })
1235 .collect()
1236 }
1237
1238 fn reassign_ids(
1239 schema: Schema,
1240 spec: UnboundPartitionSpec,
1241 sort_order: SortOrder,
1242 ) -> Result<(Schema, PartitionSpec, SortOrder)> {
1243 let previous_id_to_name = schema.field_id_to_name_map().clone();
1245 let fresh_schema = schema
1246 .into_builder()
1247 .with_schema_id(DEFAULT_SCHEMA_ID)
1248 .with_reassigned_field_ids(FIRST_FIELD_ID)
1249 .build()?;
1250
1251 let mut fresh_spec = PartitionSpecBuilder::new(fresh_schema.clone());
1253 for field in spec.fields() {
1254 let source_id = field.source_id()?;
1255 let source_field_name = previous_id_to_name.get(&source_id).ok_or_else(|| {
1256 invalid_data!(
1257 "Cannot find source column with id {} for partition column {} in schema.",
1258 source_id,
1259 field.name()
1260 )
1261 })?;
1262 fresh_spec = fresh_spec.add_partition_field(
1263 source_field_name,
1264 field.name(),
1265 field.transform(),
1266 )?;
1267 }
1268 let fresh_spec = fresh_spec.build()?;
1269
1270 let mut fresh_order = SortOrder::builder();
1272 for mut field in sort_order.fields {
1273 let source_field_name = previous_id_to_name.get(&field.source_id).ok_or_else(|| {
1274 invalid_data!(
1275 "Cannot find source column with id {} for sort column in schema.",
1276 field.source_id
1277 )
1278 })?;
1279 let new_field_id = fresh_schema
1280 .field_by_name(source_field_name)
1281 .ok_or_else(|| {
1282 Error::new(
1283 ErrorKind::Unexpected,
1284 format!(
1285 "Cannot find source column with name {source_field_name} for sort column in re-assigned schema."
1286 ),
1287 )
1288 })?.id;
1289 field.source_id = new_field_id;
1290 fresh_order.with_sort_field(field);
1291 }
1292 let fresh_sort_order = fresh_order.build(&fresh_schema)?;
1293
1294 Ok((fresh_schema, fresh_spec, fresh_sort_order))
1295 }
1296
1297 fn reuse_or_create_new_schema_id(&self, new_schema: &Schema) -> i32 {
1298 self.metadata
1299 .schemas
1300 .iter()
1301 .find_map(|(id, schema)| new_schema.is_same_schema(schema).then_some(*id))
1302 .unwrap_or_else(|| self.get_highest_schema_id() + 1)
1303 }
1304
1305 fn get_highest_schema_id(&self) -> i32 {
1306 *self
1307 .metadata
1308 .schemas
1309 .keys()
1310 .max()
1311 .unwrap_or(&self.metadata.current_schema_id)
1312 }
1313
1314 fn get_current_schema(&self) -> Result<&SchemaRef> {
1315 self.metadata
1316 .schemas
1317 .get(&self.metadata.current_schema_id)
1318 .ok_or_else(|| {
1319 invalid_data!(
1320 "Current schema with id '{}' not found in table metadata.",
1321 self.metadata.current_schema_id
1322 )
1323 })
1324 }
1325
1326 fn get_default_sort_order(&self) -> Result<SortOrderRef> {
1327 self.metadata
1328 .sort_orders
1329 .get(&self.metadata.default_sort_order_id)
1330 .cloned()
1331 .ok_or_else(|| {
1332 invalid_data!(
1333 "Default sort order with id '{}' not found in table metadata.",
1334 self.metadata.default_sort_order_id
1335 )
1336 })
1337 }
1338
1339 fn reuse_or_create_new_spec_id(&self, new_spec: &PartitionSpec) -> i32 {
1341 self.metadata
1342 .partition_specs
1343 .iter()
1344 .find_map(|(id, old_spec)| new_spec.is_compatible_with(old_spec).then_some(*id))
1345 .unwrap_or_else(|| {
1346 self.get_highest_spec_id()
1347 .map(|id| id + 1)
1348 .unwrap_or(DEFAULT_PARTITION_SPEC_ID)
1349 })
1350 }
1351
1352 fn get_highest_spec_id(&self) -> Option<i32> {
1353 self.metadata.partition_specs.keys().max().copied()
1354 }
1355
1356 fn reuse_or_create_new_sort_id(&self, new_sort_order: &SortOrder) -> i64 {
1358 if new_sort_order.is_unsorted() {
1359 return SortOrder::unsorted_order().order_id;
1360 }
1361
1362 self.metadata
1363 .sort_orders
1364 .iter()
1365 .find_map(|(id, sort_order)| {
1366 sort_order.fields.eq(&new_sort_order.fields).then_some(*id)
1367 })
1368 .unwrap_or_else(|| {
1369 self.highest_sort_order_id()
1370 .unwrap_or(SortOrder::unsorted_order().order_id)
1371 + 1
1372 })
1373 }
1374
1375 fn highest_sort_order_id(&self) -> Option<i64> {
1376 self.metadata.sort_orders.keys().max().copied()
1377 }
1378
1379 pub fn remove_schemas(mut self, schema_id_to_remove: &[i32]) -> Result<Self> {
1382 if schema_id_to_remove.contains(&self.metadata.current_schema_id) {
1383 return Err(invalid_data!("Cannot remove current schema"));
1384 }
1385
1386 if schema_id_to_remove.is_empty() {
1387 return Ok(self);
1388 }
1389
1390 let mut removed_schemas = Vec::with_capacity(schema_id_to_remove.len());
1391 self.metadata.schemas.retain(|id, _schema| {
1392 if schema_id_to_remove.contains(id) {
1393 removed_schemas.push(*id);
1394 false
1395 } else {
1396 true
1397 }
1398 });
1399
1400 self.changes.push(TableUpdate::RemoveSchemas {
1401 schema_ids: removed_schemas,
1402 });
1403
1404 Ok(self)
1405 }
1406}
1407
1408impl From<TableMetadataBuildResult> for TableMetadata {
1409 fn from(result: TableMetadataBuildResult) -> Self {
1410 result.metadata
1411 }
1412}
1413
1414#[cfg(test)]
1415mod tests {
1416 use std::fs::File;
1417 use std::io::BufReader;
1418 use std::thread::sleep;
1419
1420 use super::*;
1421 use crate::TableIdent;
1422 use crate::io::FileIO;
1423 use crate::spec::{
1424 BlobMetadata, NestedField, NullOrder, Operation, PartitionSpec, PrimitiveType, Schema,
1425 SnapshotRetention, SortDirection, SortField, StructType, Summary, TableProperties,
1426 Transform, Type, UnboundPartitionField,
1427 };
1428 use crate::table::Table;
1429 use crate::test_utils::test_runtime;
1430
1431 const TEST_LOCATION: &str = "s3://bucket/test/location";
1432 const LAST_ASSIGNED_COLUMN_ID: i32 = 3;
1433
1434 fn schema() -> Schema {
1435 Schema::builder()
1436 .with_fields(vec![
1437 NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
1438 NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
1439 NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
1440 ])
1441 .build()
1442 .unwrap()
1443 }
1444
1445 fn sort_order() -> SortOrder {
1446 let schema = schema();
1447 SortOrder::builder()
1448 .with_order_id(1)
1449 .with_sort_field(SortField {
1450 source_id: 3,
1451 transform: Transform::Bucket(4),
1452 direction: SortDirection::Descending,
1453 null_order: NullOrder::First,
1454 })
1455 .build(&schema)
1456 .unwrap()
1457 }
1458
1459 fn partition_spec() -> UnboundPartitionSpec {
1460 UnboundPartitionSpec::builder()
1461 .with_spec_id(0)
1462 .add_partition_field(
1463 UnboundPartitionField::builder()
1464 .source_ids(vec![2])
1465 .name("y")
1466 .transform(Transform::Identity)
1467 .build()
1468 .unwrap(),
1469 )
1470 .unwrap()
1471 .build()
1472 }
1473
1474 fn builder_without_changes(format_version: FormatVersion) -> TableMetadataBuilder {
1475 TableMetadataBuilder::new(
1476 schema(),
1477 partition_spec(),
1478 sort_order(),
1479 TEST_LOCATION.to_string(),
1480 format_version,
1481 HashMap::new(),
1482 )
1483 .unwrap()
1484 .build()
1485 .unwrap()
1486 .metadata
1487 .into_builder(Some(
1488 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1489 ))
1490 }
1491
1492 #[test]
1493 fn test_minimal_build() {
1494 let metadata = TableMetadataBuilder::new(
1495 schema(),
1496 partition_spec(),
1497 sort_order(),
1498 TEST_LOCATION.to_string(),
1499 FormatVersion::V1,
1500 HashMap::new(),
1501 )
1502 .unwrap()
1503 .build()
1504 .unwrap()
1505 .metadata;
1506
1507 assert_eq!(metadata.format_version, FormatVersion::V1);
1508 assert_eq!(metadata.location, TEST_LOCATION);
1509 assert_eq!(metadata.current_schema_id, 0);
1510 assert_eq!(metadata.default_spec.spec_id(), 0);
1511 assert_eq!(metadata.default_sort_order_id, 1);
1512 assert_eq!(metadata.last_partition_id, 1000);
1513 assert_eq!(metadata.last_column_id, 3);
1514 assert_eq!(metadata.snapshots.len(), 0);
1515 assert_eq!(metadata.current_snapshot_id, None);
1516 assert_eq!(metadata.refs.len(), 0);
1517 assert_eq!(metadata.properties.len(), 0);
1518 assert_eq!(metadata.metadata_log.len(), 0);
1519 assert_eq!(metadata.last_sequence_number, 0);
1520 assert_eq!(metadata.last_column_id, LAST_ASSIGNED_COLUMN_ID);
1521
1522 let _ = serde_json::to_string(&metadata).unwrap();
1524
1525 let metadata = metadata
1527 .into_builder(Some(
1528 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1529 ))
1530 .upgrade_format_version(FormatVersion::V2)
1531 .unwrap()
1532 .build()
1533 .unwrap()
1534 .metadata;
1535
1536 assert_eq!(metadata.format_version, FormatVersion::V2);
1537 let _ = serde_json::to_string(&metadata).unwrap();
1538 }
1539
1540 #[test]
1541 fn test_build_unpartitioned_unsorted() {
1542 let schema = Schema::builder().build().unwrap();
1543 let metadata = TableMetadataBuilder::new(
1544 schema.clone(),
1545 PartitionSpec::unpartition_spec(),
1546 SortOrder::unsorted_order(),
1547 TEST_LOCATION.to_string(),
1548 FormatVersion::V2,
1549 HashMap::new(),
1550 )
1551 .unwrap()
1552 .build()
1553 .unwrap()
1554 .metadata;
1555
1556 assert_eq!(metadata.format_version, FormatVersion::V2);
1557 assert_eq!(metadata.location, TEST_LOCATION);
1558 assert_eq!(metadata.current_schema_id, 0);
1559 assert_eq!(metadata.default_spec.spec_id(), 0);
1560 assert_eq!(metadata.default_sort_order_id, 0);
1561 assert_eq!(metadata.last_partition_id, UNPARTITIONED_LAST_ASSIGNED_ID);
1562 assert_eq!(metadata.last_column_id, 0);
1563 assert_eq!(metadata.snapshots.len(), 0);
1564 assert_eq!(metadata.current_snapshot_id, None);
1565 assert_eq!(metadata.refs.len(), 0);
1566 assert_eq!(metadata.properties.len(), 0);
1567 assert_eq!(metadata.metadata_log.len(), 0);
1568 assert_eq!(metadata.last_sequence_number, 0);
1569 }
1570
1571 #[test]
1572 fn test_reassigns_ids() {
1573 let schema = Schema::builder()
1574 .with_schema_id(10)
1575 .with_fields(vec![
1576 NestedField::required(11, "a", Type::Primitive(PrimitiveType::Long)).into(),
1577 NestedField::required(12, "b", Type::Primitive(PrimitiveType::Long)).into(),
1578 NestedField::required(
1579 13,
1580 "struct",
1581 Type::Struct(StructType::new(vec![
1582 NestedField::required(14, "nested", Type::Primitive(PrimitiveType::Long))
1583 .into(),
1584 ])),
1585 )
1586 .into(),
1587 NestedField::required(15, "c", Type::Primitive(PrimitiveType::Long)).into(),
1588 ])
1589 .build()
1590 .unwrap();
1591 let spec = PartitionSpec::builder(schema.clone())
1592 .with_spec_id(20)
1593 .add_partition_field("a", "a", Transform::Identity)
1594 .unwrap()
1595 .add_partition_field("struct.nested", "nested_partition", Transform::Identity)
1596 .unwrap()
1597 .build()
1598 .unwrap();
1599 let sort_order = SortOrder::builder()
1600 .with_fields(vec![SortField {
1601 source_id: 11,
1602 transform: Transform::Identity,
1603 direction: SortDirection::Ascending,
1604 null_order: NullOrder::First,
1605 }])
1606 .with_order_id(10)
1607 .build(&schema)
1608 .unwrap();
1609
1610 let (fresh_schema, fresh_spec, fresh_sort_order) =
1611 TableMetadataBuilder::reassign_ids(schema, spec.into_unbound(), sort_order).unwrap();
1612
1613 let expected_schema = Schema::builder()
1614 .with_fields(vec![
1615 NestedField::required(1, "a", Type::Primitive(PrimitiveType::Long)).into(),
1616 NestedField::required(2, "b", Type::Primitive(PrimitiveType::Long)).into(),
1617 NestedField::required(
1618 3,
1619 "struct",
1620 Type::Struct(StructType::new(vec![
1621 NestedField::required(5, "nested", Type::Primitive(PrimitiveType::Long))
1622 .into(),
1623 ])),
1624 )
1625 .into(),
1626 NestedField::required(4, "c", Type::Primitive(PrimitiveType::Long)).into(),
1627 ])
1628 .build()
1629 .unwrap();
1630
1631 let expected_spec = PartitionSpec::builder(expected_schema.clone())
1632 .with_spec_id(0)
1633 .add_partition_field("a", "a", Transform::Identity)
1634 .unwrap()
1635 .add_partition_field("struct.nested", "nested_partition", Transform::Identity)
1636 .unwrap()
1637 .build()
1638 .unwrap();
1639
1640 let expected_sort_order = SortOrder::builder()
1641 .with_fields(vec![SortField {
1642 source_id: 1,
1643 transform: Transform::Identity,
1644 direction: SortDirection::Ascending,
1645 null_order: NullOrder::First,
1646 }])
1647 .with_order_id(1)
1648 .build(&expected_schema)
1649 .unwrap();
1650
1651 assert_eq!(fresh_schema, expected_schema);
1652 assert_eq!(fresh_spec, expected_spec);
1653 assert_eq!(fresh_sort_order, expected_sort_order);
1654 }
1655
1656 #[test]
1657 fn test_ids_are_reassigned_for_new_metadata() {
1658 let schema = schema().into_builder().with_schema_id(10).build().unwrap();
1659
1660 let metadata = TableMetadataBuilder::new(
1661 schema,
1662 partition_spec(),
1663 sort_order(),
1664 TEST_LOCATION.to_string(),
1665 FormatVersion::V1,
1666 HashMap::new(),
1667 )
1668 .unwrap()
1669 .build()
1670 .unwrap()
1671 .metadata;
1672
1673 assert_eq!(metadata.current_schema_id, 0);
1674 assert_eq!(metadata.current_schema().schema_id(), 0);
1675 }
1676
1677 #[test]
1678 fn test_new_metadata_changes() {
1679 let changes = TableMetadataBuilder::new(
1680 schema(),
1681 partition_spec(),
1682 sort_order(),
1683 TEST_LOCATION.to_string(),
1684 FormatVersion::V1,
1685 HashMap::from_iter(vec![("property 1".to_string(), "value 1".to_string())]),
1686 )
1687 .unwrap()
1688 .build()
1689 .unwrap()
1690 .changes;
1691
1692 pretty_assertions::assert_eq!(changes, vec![
1693 TableUpdate::SetLocation {
1694 location: TEST_LOCATION.to_string()
1695 },
1696 TableUpdate::AddSchema { schema: schema() },
1697 TableUpdate::SetCurrentSchema { schema_id: -1 },
1698 TableUpdate::AddSpec {
1699 spec: PartitionSpec::builder(schema())
1702 .with_spec_id(0)
1703 .add_unbound_field(
1704 UnboundPartitionField::builder()
1705 .source_ids(vec![2])
1706 .field_id(1000)
1707 .name("y".to_string())
1708 .transform(Transform::Identity)
1709 .build()
1710 .unwrap()
1711 )
1712 .unwrap()
1713 .build()
1714 .unwrap()
1715 .into_unbound(),
1716 },
1717 TableUpdate::SetDefaultSpec { spec_id: -1 },
1718 TableUpdate::AddSortOrder {
1719 sort_order: sort_order(),
1720 },
1721 TableUpdate::SetDefaultSortOrder { sort_order_id: -1 },
1722 TableUpdate::SetProperties {
1723 updates: HashMap::from_iter(vec![(
1724 "property 1".to_string(),
1725 "value 1".to_string()
1726 )]),
1727 }
1728 ]);
1729 }
1730
1731 #[test]
1732 fn test_new_metadata_changes_unpartitioned_unsorted() {
1733 let schema = Schema::builder().build().unwrap();
1734 let changes = TableMetadataBuilder::new(
1735 schema.clone(),
1736 PartitionSpec::unpartition_spec().into_unbound(),
1737 SortOrder::unsorted_order(),
1738 TEST_LOCATION.to_string(),
1739 FormatVersion::V1,
1740 HashMap::new(),
1741 )
1742 .unwrap()
1743 .build()
1744 .unwrap()
1745 .changes;
1746
1747 pretty_assertions::assert_eq!(changes, vec![
1748 TableUpdate::SetLocation {
1749 location: TEST_LOCATION.to_string()
1750 },
1751 TableUpdate::AddSchema {
1752 schema: Schema::builder().build().unwrap(),
1753 },
1754 TableUpdate::SetCurrentSchema { schema_id: -1 },
1755 TableUpdate::AddSpec {
1756 spec: PartitionSpec::builder(schema)
1759 .with_spec_id(0)
1760 .build()
1761 .unwrap()
1762 .into_unbound(),
1763 },
1764 TableUpdate::SetDefaultSpec { spec_id: -1 },
1765 TableUpdate::AddSortOrder {
1766 sort_order: SortOrder::unsorted_order(),
1767 },
1768 TableUpdate::SetDefaultSortOrder { sort_order_id: -1 },
1769 ]);
1770 }
1771
1772 #[test]
1773 fn test_add_partition_spec() {
1774 let builder = builder_without_changes(FormatVersion::V2);
1775
1776 let added_spec = UnboundPartitionSpec::builder()
1777 .with_spec_id(10)
1778 .add_partition_fields(vec![
1779 UnboundPartitionField::builder()
1781 .source_ids(vec![2])
1782 .field_id(1000)
1783 .name("y".to_string())
1784 .transform(Transform::Identity)
1785 .build()
1786 .unwrap(),
1787 UnboundPartitionField::builder()
1789 .source_ids(vec![3])
1790 .name("z".to_string())
1791 .transform(Transform::Identity)
1792 .build()
1793 .unwrap(),
1794 ])
1795 .unwrap()
1796 .build();
1797
1798 let build_result = builder
1799 .add_partition_spec(added_spec.clone())
1800 .unwrap()
1801 .build()
1802 .unwrap();
1803
1804 let expected_change = added_spec.with_spec_id(1);
1806 let expected_spec = PartitionSpec::builder(schema())
1807 .with_spec_id(1)
1808 .add_unbound_field(
1809 UnboundPartitionField::builder()
1810 .source_ids(vec![2])
1811 .field_id(1000)
1812 .name("y".to_string())
1813 .transform(Transform::Identity)
1814 .build()
1815 .unwrap(),
1816 )
1817 .unwrap()
1818 .add_unbound_field(
1819 UnboundPartitionField::builder()
1820 .source_ids(vec![3])
1821 .field_id(1001)
1822 .name("z".to_string())
1823 .transform(Transform::Identity)
1824 .build()
1825 .unwrap(),
1826 )
1827 .unwrap()
1828 .build()
1829 .unwrap();
1830
1831 assert_eq!(build_result.changes.len(), 1);
1832 assert_eq!(
1833 build_result.metadata.partition_spec_by_id(1),
1834 Some(&Arc::new(expected_spec))
1835 );
1836 assert_eq!(build_result.metadata.default_spec.spec_id(), 0);
1837 assert_eq!(build_result.metadata.last_partition_id, 1001);
1838 pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSpec {
1839 spec: expected_change
1840 });
1841
1842 let build_result = build_result
1844 .metadata
1845 .into_builder(Some(
1846 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1847 ))
1848 .remove_partition_specs(&[1])
1849 .unwrap()
1850 .build()
1851 .unwrap();
1852
1853 assert_eq!(build_result.changes.len(), 1);
1854 assert_eq!(build_result.metadata.partition_specs.len(), 1);
1855 assert!(build_result.metadata.partition_spec_by_id(1).is_none());
1856 }
1857
1858 #[test]
1859 fn test_set_default_partition_spec() {
1860 let builder = builder_without_changes(FormatVersion::V2);
1861 let schema = builder.get_current_schema().unwrap().clone();
1862 let added_spec = UnboundPartitionSpec::builder()
1863 .with_spec_id(10)
1864 .add_partition_field(
1865 UnboundPartitionField::builder()
1866 .source_ids(vec![1])
1867 .name("y_bucket[2]")
1868 .transform(Transform::Bucket(2))
1869 .build()
1870 .unwrap(),
1871 )
1872 .unwrap()
1873 .build();
1874
1875 let build_result = builder
1876 .add_partition_spec(added_spec.clone())
1877 .unwrap()
1878 .set_default_partition_spec(-1)
1879 .unwrap()
1880 .build()
1881 .unwrap();
1882
1883 let expected_spec = PartitionSpec::builder(schema)
1884 .with_spec_id(1)
1885 .add_unbound_field(
1886 UnboundPartitionField::builder()
1887 .source_ids(vec![1])
1888 .field_id(1001)
1889 .name("y_bucket[2]".to_string())
1890 .transform(Transform::Bucket(2))
1891 .build()
1892 .unwrap(),
1893 )
1894 .unwrap()
1895 .build()
1896 .unwrap();
1897
1898 assert_eq!(build_result.changes.len(), 2);
1899 assert_eq!(build_result.metadata.default_spec, Arc::new(expected_spec));
1900 assert_eq!(build_result.changes, vec![
1901 TableUpdate::AddSpec {
1902 spec: added_spec.with_spec_id(1)
1904 },
1905 TableUpdate::SetDefaultSpec { spec_id: -1 }
1906 ]);
1907 }
1908
1909 #[test]
1910 fn test_set_existing_default_partition_spec() {
1911 let builder = builder_without_changes(FormatVersion::V2);
1912 let unbound_spec = UnboundPartitionSpec::builder().with_spec_id(1).build();
1914 let build_result = builder
1915 .add_partition_spec(unbound_spec.clone())
1916 .unwrap()
1917 .set_default_partition_spec(-1)
1918 .unwrap()
1919 .build()
1920 .unwrap();
1921
1922 assert_eq!(build_result.changes.len(), 2);
1923 assert_eq!(build_result.changes[0], TableUpdate::AddSpec {
1924 spec: unbound_spec.clone()
1925 });
1926 assert_eq!(build_result.changes[1], TableUpdate::SetDefaultSpec {
1927 spec_id: -1
1928 });
1929 assert_eq!(
1930 build_result.metadata.default_spec,
1931 Arc::new(
1932 unbound_spec
1933 .bind(build_result.metadata.current_schema().clone())
1934 .unwrap()
1935 )
1936 );
1937
1938 let build_result = build_result
1940 .metadata
1941 .into_builder(Some(
1942 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
1943 ))
1944 .set_default_partition_spec(0)
1945 .unwrap()
1946 .build()
1947 .unwrap();
1948
1949 assert_eq!(build_result.changes.len(), 1);
1950 assert_eq!(build_result.changes[0], TableUpdate::SetDefaultSpec {
1951 spec_id: 0
1952 });
1953 assert_eq!(
1954 build_result.metadata.default_spec,
1955 Arc::new(
1956 partition_spec()
1957 .bind(build_result.metadata.current_schema().clone())
1958 .unwrap()
1959 )
1960 );
1961 }
1962
1963 #[test]
1964 fn test_add_sort_order() {
1965 let builder = builder_without_changes(FormatVersion::V2);
1966
1967 let added_sort_order = SortOrder::builder()
1968 .with_order_id(10)
1969 .with_fields(vec![SortField {
1970 source_id: 1,
1971 transform: Transform::Identity,
1972 direction: SortDirection::Ascending,
1973 null_order: NullOrder::First,
1974 }])
1975 .build(&schema())
1976 .unwrap();
1977
1978 let build_result = builder
1979 .add_sort_order(added_sort_order.clone())
1980 .unwrap()
1981 .build()
1982 .unwrap();
1983
1984 let expected_sort_order = added_sort_order.with_order_id(2);
1985
1986 assert_eq!(build_result.changes.len(), 1);
1987 assert_eq!(build_result.metadata.sort_orders.keys().max(), Some(&2));
1988 pretty_assertions::assert_eq!(
1989 build_result.metadata.sort_order_by_id(2),
1990 Some(&Arc::new(expected_sort_order.clone()))
1991 );
1992 pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSortOrder {
1993 sort_order: expected_sort_order
1994 });
1995 }
1996
1997 #[test]
1998 fn test_add_compatible_schema() {
1999 let builder = builder_without_changes(FormatVersion::V2);
2000
2001 let added_schema = Schema::builder()
2002 .with_schema_id(1)
2003 .with_fields(vec![
2004 NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
2005 NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
2006 NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
2007 NestedField::required(4, "a", Type::Primitive(PrimitiveType::Long)).into(),
2008 ])
2009 .build()
2010 .unwrap();
2011
2012 let build_result = builder
2013 .add_current_schema(added_schema.clone())
2014 .unwrap()
2015 .build()
2016 .unwrap();
2017
2018 assert_eq!(build_result.changes.len(), 2);
2019 assert_eq!(build_result.metadata.schemas.keys().max(), Some(&1));
2020 pretty_assertions::assert_eq!(
2021 build_result.metadata.schema_by_id(1),
2022 Some(&Arc::new(added_schema.clone()))
2023 );
2024 pretty_assertions::assert_eq!(build_result.changes[0], TableUpdate::AddSchema {
2025 schema: added_schema
2026 });
2027 assert_eq!(build_result.changes[1], TableUpdate::SetCurrentSchema {
2028 schema_id: -1
2029 });
2030 }
2031
2032 #[test]
2033 fn test_set_current_schema_change_is_minus_one_if_schema_was_added_in_this_change() {
2034 let builder = builder_without_changes(FormatVersion::V2);
2035
2036 let added_schema = Schema::builder()
2037 .with_schema_id(1)
2038 .with_fields(vec![
2039 NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
2040 NestedField::required(2, "y", Type::Primitive(PrimitiveType::Long)).into(),
2041 NestedField::required(3, "z", Type::Primitive(PrimitiveType::Long)).into(),
2042 NestedField::required(4, "a", Type::Primitive(PrimitiveType::Long)).into(),
2043 ])
2044 .build()
2045 .unwrap();
2046
2047 let build_result = builder
2048 .add_schema(added_schema.clone())
2049 .unwrap()
2050 .set_current_schema(1)
2051 .unwrap()
2052 .build()
2053 .unwrap();
2054
2055 assert_eq!(build_result.changes.len(), 2);
2056 assert_eq!(build_result.changes[1], TableUpdate::SetCurrentSchema {
2057 schema_id: -1
2058 });
2059 }
2060
2061 #[test]
2062 fn test_no_metadata_log_for_create_table() {
2063 let build_result = TableMetadataBuilder::new(
2064 schema(),
2065 partition_spec(),
2066 sort_order(),
2067 TEST_LOCATION.to_string(),
2068 FormatVersion::V2,
2069 HashMap::new(),
2070 )
2071 .unwrap()
2072 .build()
2073 .unwrap();
2074
2075 assert_eq!(build_result.metadata.metadata_log.len(), 0);
2076 }
2077
2078 #[test]
2079 fn test_table_properties_view_reflects_metadata_updates() {
2080 let property = TableProperties::PROPERTY_COMMIT_NUM_RETRIES.to_string();
2081 let metadata = builder_without_changes(FormatVersion::V2)
2082 .set_properties(HashMap::from([(property.clone(), "7".to_string())]))
2083 .unwrap()
2084 .build()
2085 .unwrap()
2086 .metadata;
2087
2088 assert_eq!(metadata.table_properties().commit_num_retries().unwrap(), 7);
2089
2090 let metadata = metadata
2091 .into_builder(None)
2092 .remove_properties(&[property])
2093 .unwrap()
2094 .build()
2095 .unwrap()
2096 .metadata;
2097
2098 assert_eq!(
2099 metadata.table_properties().commit_num_retries().unwrap(),
2100 TableProperties::PROPERTY_COMMIT_NUM_RETRIES_DEFAULT
2101 );
2102 }
2103
2104 #[test]
2105 fn test_no_metadata_log_entry_for_no_previous_location() {
2106 let metadata = builder_without_changes(FormatVersion::V2)
2108 .build()
2109 .unwrap()
2110 .metadata;
2111 assert_eq!(metadata.metadata_log.len(), 1);
2112
2113 let build_result = metadata
2114 .into_builder(None)
2115 .set_properties(HashMap::from_iter(vec![(
2116 "foo".to_string(),
2117 "bar".to_string(),
2118 )]))
2119 .unwrap()
2120 .build()
2121 .unwrap();
2122
2123 assert_eq!(build_result.metadata.metadata_log.len(), 1);
2124 }
2125
2126 #[test]
2127 fn test_from_metadata_generates_metadata_log() {
2128 let metadata_path = "s3://bucket/test/location/metadata/metadata1.json";
2129 let builder = TableMetadataBuilder::new(
2130 schema(),
2131 partition_spec(),
2132 sort_order(),
2133 TEST_LOCATION.to_string(),
2134 FormatVersion::V2,
2135 HashMap::new(),
2136 )
2137 .unwrap()
2138 .build()
2139 .unwrap()
2140 .metadata
2141 .into_builder(Some(metadata_path.to_string()));
2142
2143 let builder = builder
2144 .add_default_sort_order(SortOrder::unsorted_order())
2145 .unwrap();
2146
2147 let build_result = builder.build().unwrap();
2148
2149 assert_eq!(build_result.metadata.metadata_log.len(), 1);
2150 assert_eq!(
2151 build_result.metadata.metadata_log[0].metadata_file,
2152 metadata_path
2153 );
2154 }
2155
2156 #[test]
2157 fn test_set_ref() {
2158 let builder = builder_without_changes(FormatVersion::V2);
2159
2160 let snapshot = Snapshot::builder()
2161 .with_snapshot_id(1)
2162 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2163 .with_sequence_number(0)
2164 .with_schema_id(0)
2165 .with_manifest_list("/snap-1.avro")
2166 .with_summary(Summary {
2167 operation: Operation::Append,
2168 additional_properties: HashMap::from_iter(vec![
2169 (
2170 "spark.app.id".to_string(),
2171 "local-1662532784305".to_string(),
2172 ),
2173 ("added-data-files".to_string(), "4".to_string()),
2174 ("added-records".to_string(), "4".to_string()),
2175 ("added-files-size".to_string(), "6001".to_string()),
2176 ]),
2177 })
2178 .build();
2179
2180 let builder = builder.add_snapshot(snapshot.clone()).unwrap();
2181
2182 assert!(
2183 builder
2184 .clone()
2185 .set_ref(MAIN_BRANCH, SnapshotReference {
2186 snapshot_id: 10,
2187 retention: SnapshotRetention::Branch {
2188 min_snapshots_to_keep: Some(10),
2189 max_snapshot_age_ms: None,
2190 max_ref_age_ms: None,
2191 },
2192 })
2193 .unwrap_err()
2194 .to_string()
2195 .contains("Cannot set 'main' to unknown snapshot: '10'")
2196 );
2197
2198 let build_result = builder
2199 .set_ref(MAIN_BRANCH, SnapshotReference {
2200 snapshot_id: 1,
2201 retention: SnapshotRetention::Branch {
2202 min_snapshots_to_keep: Some(10),
2203 max_snapshot_age_ms: None,
2204 max_ref_age_ms: None,
2205 },
2206 })
2207 .unwrap()
2208 .build()
2209 .unwrap();
2210 assert_eq!(build_result.metadata.snapshots.len(), 1);
2211 assert_eq!(
2212 build_result.metadata.snapshot_by_id(1),
2213 Some(&Arc::new(snapshot.clone()))
2214 );
2215 assert_eq!(build_result.metadata.snapshot_log, vec![SnapshotLog {
2216 snapshot_id: 1,
2217 timestamp_ms: snapshot.timestamp_ms()
2218 }])
2219 }
2220
2221 #[test]
2222 fn test_snapshot_log_skips_intermediates() {
2223 let builder = builder_without_changes(FormatVersion::V2);
2224
2225 let snapshot_1 = Snapshot::builder()
2226 .with_snapshot_id(1)
2227 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2228 .with_sequence_number(0)
2229 .with_schema_id(0)
2230 .with_manifest_list("/snap-1.avro")
2231 .with_summary(Summary {
2232 operation: Operation::Append,
2233 additional_properties: HashMap::from_iter(vec![
2234 (
2235 "spark.app.id".to_string(),
2236 "local-1662532784305".to_string(),
2237 ),
2238 ("added-data-files".to_string(), "4".to_string()),
2239 ("added-records".to_string(), "4".to_string()),
2240 ("added-files-size".to_string(), "6001".to_string()),
2241 ]),
2242 })
2243 .build();
2244
2245 let snapshot_2 = Snapshot::builder()
2246 .with_snapshot_id(2)
2247 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2248 .with_sequence_number(0)
2249 .with_schema_id(0)
2250 .with_manifest_list("/snap-1.avro")
2251 .with_summary(Summary {
2252 operation: Operation::Append,
2253 additional_properties: HashMap::from_iter(vec![
2254 (
2255 "spark.app.id".to_string(),
2256 "local-1662532784305".to_string(),
2257 ),
2258 ("added-data-files".to_string(), "4".to_string()),
2259 ("added-records".to_string(), "4".to_string()),
2260 ("added-files-size".to_string(), "6001".to_string()),
2261 ]),
2262 })
2263 .build();
2264
2265 let result = builder
2266 .add_snapshot(snapshot_1)
2267 .unwrap()
2268 .set_ref(MAIN_BRANCH, SnapshotReference {
2269 snapshot_id: 1,
2270 retention: SnapshotRetention::Branch {
2271 min_snapshots_to_keep: Some(10),
2272 max_snapshot_age_ms: None,
2273 max_ref_age_ms: None,
2274 },
2275 })
2276 .unwrap()
2277 .set_branch_snapshot(snapshot_2.clone(), MAIN_BRANCH)
2278 .unwrap()
2279 .build()
2280 .unwrap();
2281
2282 assert_eq!(result.metadata.snapshot_log, vec![SnapshotLog {
2283 snapshot_id: 2,
2284 timestamp_ms: snapshot_2.timestamp_ms()
2285 }]);
2286 assert_eq!(result.metadata.current_snapshot().unwrap().snapshot_id(), 2);
2287 }
2288
2289 #[test]
2290 fn test_remove_main_ref_keeps_snapshot_log() {
2291 let builder = builder_without_changes(FormatVersion::V2);
2292
2293 let snapshot = Snapshot::builder()
2294 .with_snapshot_id(1)
2295 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2296 .with_sequence_number(0)
2297 .with_schema_id(0)
2298 .with_manifest_list("/snap-1.avro")
2299 .with_summary(Summary {
2300 operation: Operation::Append,
2301 additional_properties: HashMap::from_iter(vec![
2302 (
2303 "spark.app.id".to_string(),
2304 "local-1662532784305".to_string(),
2305 ),
2306 ("added-data-files".to_string(), "4".to_string()),
2307 ("added-records".to_string(), "4".to_string()),
2308 ("added-files-size".to_string(), "6001".to_string()),
2309 ]),
2310 })
2311 .build();
2312
2313 let result = builder
2314 .add_snapshot(snapshot.clone())
2315 .unwrap()
2316 .set_ref(MAIN_BRANCH, SnapshotReference {
2317 snapshot_id: 1,
2318 retention: SnapshotRetention::Branch {
2319 min_snapshots_to_keep: Some(10),
2320 max_snapshot_age_ms: None,
2321 max_ref_age_ms: None,
2322 },
2323 })
2324 .unwrap()
2325 .build()
2326 .unwrap();
2327
2328 assert_eq!(result.metadata.snapshot_log.len(), 1);
2330 assert_eq!(result.metadata.snapshot_log[0].snapshot_id, 1);
2331 assert_eq!(result.metadata.current_snapshot_id, Some(1));
2332
2333 let result_after_remove = result
2335 .metadata
2336 .into_builder(Some(
2337 "s3://bucket/test/location/metadata/metadata2.json".to_string(),
2338 ))
2339 .remove_ref(MAIN_BRANCH)
2340 .build()
2341 .unwrap();
2342
2343 assert_eq!(result_after_remove.metadata.snapshot_log.len(), 1);
2345 assert_eq!(result_after_remove.metadata.snapshot_log[0].snapshot_id, 1);
2346 assert_eq!(result_after_remove.metadata.current_snapshot_id, None);
2347 assert_eq!(result_after_remove.changes.len(), 1);
2348 assert_eq!(
2349 result_after_remove.changes[0],
2350 TableUpdate::RemoveSnapshotRef {
2351 ref_name: MAIN_BRANCH.to_string()
2352 }
2353 );
2354 }
2355
2356 #[test]
2357 fn test_set_branch_snapshot_creates_branch_if_not_exists() {
2358 let builder = builder_without_changes(FormatVersion::V2);
2359
2360 let snapshot = Snapshot::builder()
2361 .with_snapshot_id(2)
2362 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2363 .with_sequence_number(0)
2364 .with_schema_id(0)
2365 .with_manifest_list("/snap-1.avro")
2366 .with_summary(Summary {
2367 operation: Operation::Append,
2368 additional_properties: HashMap::new(),
2369 })
2370 .build();
2371
2372 let build_result = builder
2373 .set_branch_snapshot(snapshot.clone(), "new_branch")
2374 .unwrap()
2375 .build()
2376 .unwrap();
2377
2378 let reference = SnapshotReference {
2379 snapshot_id: 2,
2380 retention: SnapshotRetention::Branch {
2381 min_snapshots_to_keep: None,
2382 max_snapshot_age_ms: None,
2383 max_ref_age_ms: None,
2384 },
2385 };
2386
2387 assert_eq!(build_result.metadata.refs.len(), 1);
2388 assert_eq!(
2389 build_result.metadata.refs.get("new_branch"),
2390 Some(&reference)
2391 );
2392 assert_eq!(build_result.changes, vec![
2393 TableUpdate::AddSnapshot { snapshot },
2394 TableUpdate::SetSnapshotRef {
2395 ref_name: "new_branch".to_string(),
2396 reference
2397 }
2398 ]);
2399 }
2400
2401 #[test]
2402 fn test_cannot_add_duplicate_snapshot_id() {
2403 let builder = builder_without_changes(FormatVersion::V2);
2404
2405 let snapshot = Snapshot::builder()
2406 .with_snapshot_id(2)
2407 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2408 .with_sequence_number(0)
2409 .with_schema_id(0)
2410 .with_manifest_list("/snap-1.avro")
2411 .with_summary(Summary {
2412 operation: Operation::Append,
2413 additional_properties: HashMap::from_iter(vec![
2414 (
2415 "spark.app.id".to_string(),
2416 "local-1662532784305".to_string(),
2417 ),
2418 ("added-data-files".to_string(), "4".to_string()),
2419 ("added-records".to_string(), "4".to_string()),
2420 ("added-files-size".to_string(), "6001".to_string()),
2421 ]),
2422 })
2423 .build();
2424
2425 let builder = builder.add_snapshot(snapshot.clone()).unwrap();
2426 builder.add_snapshot(snapshot).unwrap_err();
2427 }
2428
2429 #[test]
2430 fn test_add_incompatible_current_schema_fails() {
2431 let builder = builder_without_changes(FormatVersion::V2);
2432
2433 let added_schema = Schema::builder()
2434 .with_schema_id(1)
2435 .with_fields(vec![])
2436 .build()
2437 .unwrap();
2438
2439 let err = builder
2440 .add_current_schema(added_schema)
2441 .unwrap()
2442 .build()
2443 .unwrap_err();
2444
2445 assert!(
2446 err.to_string()
2447 .contains("Cannot find partition source field")
2448 );
2449 }
2450
2451 #[test]
2452 fn test_add_partition_spec_for_v1_requires_sequential_ids() {
2453 let builder = builder_without_changes(FormatVersion::V1);
2454
2455 let added_spec = UnboundPartitionSpec::builder()
2456 .with_spec_id(10)
2457 .add_partition_fields(vec![
2458 UnboundPartitionField::builder()
2459 .source_ids(vec![2])
2460 .field_id(1000)
2461 .name("y".to_string())
2462 .transform(Transform::Identity)
2463 .build()
2464 .unwrap(),
2465 UnboundPartitionField::builder()
2466 .source_ids(vec![3])
2467 .field_id(1002)
2468 .name("z".to_string())
2469 .transform(Transform::Identity)
2470 .build()
2471 .unwrap(),
2472 ])
2473 .unwrap()
2474 .build();
2475
2476 let err = builder.add_partition_spec(added_spec).unwrap_err();
2477 assert!(err.to_string().contains(
2478 "Cannot add partition spec with non-sequential field ids to format version 1 table"
2479 ));
2480 }
2481
2482 #[test]
2483 fn test_add_partition_spec_accepts_void_fields_of_removed_fields() {
2484 let builder = builder_without_changes(FormatVersion::V1);
2485 let void_field = |field_id: i32, name: &str| {
2486 UnboundPartitionField::builder()
2487 .source_ids(vec![2])
2488 .field_id(field_id)
2489 .name(name)
2490 .transform(Transform::Void)
2491 .build()
2492 .unwrap()
2493 };
2494 let added_spec = UnboundPartitionSpec {
2495 spec_id: None,
2496 fields: vec![void_field(1000, "y"), void_field(1001, "y_null")],
2497 };
2498
2499 let metadata = builder
2500 .add_default_partition_spec(added_spec.clone())
2501 .unwrap()
2502 .build()
2503 .unwrap()
2504 .metadata;
2505
2506 let expected = added_spec.with_spec_id(1).bind(schema()).unwrap();
2507 assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
2508 assert_eq!(metadata.last_partition_id, 1001);
2509 }
2510
2511 #[test]
2512 fn test_add_partition_spec_accepts_two_time_granularities_of_one_column() {
2513 let schema = Schema::builder()
2514 .with_fields(vec![
2515 NestedField::required(1, "ts", Type::Primitive(PrimitiveType::Timestamptz)).into(),
2516 ])
2517 .build()
2518 .unwrap();
2519 let time_field = |field_id: i32, name: &str, transform: Transform| {
2520 UnboundPartitionField::builder()
2521 .source_ids(vec![1])
2522 .field_id(field_id)
2523 .name(name)
2524 .transform(transform)
2525 .build()
2526 .unwrap()
2527 };
2528 let added_spec = UnboundPartitionSpec {
2529 spec_id: None,
2530 fields: vec![
2531 time_field(1000, "ts_day", Transform::Day),
2532 time_field(1001, "ts_hour", Transform::Hour),
2533 ],
2534 };
2535 let metadata = TableMetadataBuilder::new(
2536 schema.clone(),
2537 PartitionSpec::unpartition_spec(),
2538 SortOrder::unsorted_order(),
2539 TEST_LOCATION.to_string(),
2540 FormatVersion::V2,
2541 HashMap::new(),
2542 )
2543 .unwrap()
2544 .build()
2545 .unwrap()
2546 .metadata;
2547
2548 let metadata = metadata
2549 .into_builder(None)
2550 .add_default_partition_spec(added_spec.clone())
2551 .unwrap()
2552 .build()
2553 .unwrap()
2554 .metadata;
2555 let metadata = metadata
2557 .into_builder(None)
2558 .set_properties(HashMap::from([("key".to_string(), "value".to_string())]))
2559 .unwrap()
2560 .build()
2561 .unwrap()
2562 .metadata;
2563
2564 let expected = added_spec.with_spec_id(1).bind(schema).unwrap();
2565 assert_eq!(metadata.default_partition_spec().as_ref(), &expected);
2566 }
2567
2568 #[test]
2569 fn test_expire_metadata_log() {
2570 let builder = builder_without_changes(FormatVersion::V2);
2571 let metadata = builder
2572 .set_properties(HashMap::from_iter(vec![(
2573 TableProperties::PROPERTY_METADATA_PREVIOUS_VERSIONS_MAX.to_string(),
2574 "2".to_string(),
2575 )]))
2576 .unwrap()
2577 .build()
2578 .unwrap();
2579 assert_eq!(metadata.metadata.metadata_log.len(), 1);
2580 assert_eq!(metadata.expired_metadata_logs.len(), 0);
2581
2582 let metadata = metadata
2583 .metadata
2584 .into_builder(Some("path2".to_string()))
2585 .set_properties(HashMap::from_iter(vec![(
2586 "change_nr".to_string(),
2587 "1".to_string(),
2588 )]))
2589 .unwrap()
2590 .build()
2591 .unwrap();
2592
2593 assert_eq!(metadata.metadata.metadata_log.len(), 2);
2594 assert_eq!(metadata.expired_metadata_logs.len(), 0);
2595
2596 let metadata = metadata
2597 .metadata
2598 .into_builder(Some("path2".to_string()))
2599 .set_properties(HashMap::from_iter(vec![(
2600 "change_nr".to_string(),
2601 "2".to_string(),
2602 )]))
2603 .unwrap()
2604 .build()
2605 .unwrap();
2606 assert_eq!(metadata.metadata.metadata_log.len(), 2);
2607 assert_eq!(metadata.expired_metadata_logs.len(), 1);
2608 }
2609
2610 #[test]
2611 fn test_v2_sequence_number_cannot_decrease() {
2612 let builder = builder_without_changes(FormatVersion::V2);
2613
2614 let snapshot = Snapshot::builder()
2615 .with_snapshot_id(1)
2616 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2617 .with_sequence_number(1)
2618 .with_schema_id(0)
2619 .with_manifest_list("/snap-1")
2620 .with_summary(Summary {
2621 operation: Operation::Append,
2622 additional_properties: HashMap::new(),
2623 })
2624 .build();
2625
2626 let builder = builder
2627 .add_snapshot(snapshot.clone())
2628 .unwrap()
2629 .set_ref(MAIN_BRANCH, SnapshotReference {
2630 snapshot_id: 1,
2631 retention: SnapshotRetention::Branch {
2632 min_snapshots_to_keep: Some(10),
2633 max_snapshot_age_ms: None,
2634 max_ref_age_ms: None,
2635 },
2636 })
2637 .unwrap();
2638
2639 let snapshot = Snapshot::builder()
2640 .with_snapshot_id(2)
2641 .with_timestamp_ms(builder.metadata.last_updated_ms + 1)
2642 .with_sequence_number(0)
2643 .with_schema_id(0)
2644 .with_manifest_list("/snap-0")
2645 .with_parent_snapshot_id(Some(1))
2646 .with_summary(Summary {
2647 operation: Operation::Append,
2648 additional_properties: HashMap::new(),
2649 })
2650 .build();
2651
2652 let err = builder
2653 .set_branch_snapshot(snapshot, MAIN_BRANCH)
2654 .unwrap_err();
2655 assert!(
2656 err.to_string()
2657 .contains("Cannot add snapshot with sequence number")
2658 );
2659 }
2660
2661 #[test]
2662 fn test_default_spec_cannot_be_removed() {
2663 let builder = builder_without_changes(FormatVersion::V2);
2664
2665 builder.remove_partition_specs(&[0]).unwrap_err();
2666 }
2667
2668 #[test]
2669 fn test_statistics() {
2670 let builder = builder_without_changes(FormatVersion::V2);
2671
2672 let statistics = StatisticsFile {
2673 snapshot_id: 3055729675574597004,
2674 statistics_path: "s3://a/b/stats.puffin".to_string(),
2675 file_size_in_bytes: 413,
2676 file_footer_size_in_bytes: 42,
2677 key_metadata: None,
2678 blob_metadata: vec![BlobMetadata {
2679 snapshot_id: 3055729675574597004,
2680 sequence_number: 1,
2681 fields: vec![1],
2682 r#type: "ndv".to_string(),
2683 properties: HashMap::new(),
2684 }],
2685 };
2686 let build_result = builder.set_statistics(statistics.clone()).build().unwrap();
2687
2688 assert_eq!(
2689 build_result.metadata.statistics,
2690 HashMap::from_iter(vec![(3055729675574597004, statistics.clone())])
2691 );
2692 assert_eq!(build_result.changes, vec![TableUpdate::SetStatistics {
2693 statistics: statistics.clone()
2694 }]);
2695
2696 let builder = build_result.metadata.into_builder(None);
2698 let build_result = builder
2699 .remove_statistics(statistics.snapshot_id)
2700 .build()
2701 .unwrap();
2702
2703 assert_eq!(build_result.metadata.statistics.len(), 0);
2704 assert_eq!(build_result.changes, vec![TableUpdate::RemoveStatistics {
2705 snapshot_id: statistics.snapshot_id
2706 }]);
2707
2708 let builder = build_result.metadata.into_builder(None);
2710 let build_result = builder
2711 .remove_statistics(statistics.snapshot_id)
2712 .build()
2713 .unwrap();
2714 assert_eq!(build_result.metadata.statistics.len(), 0);
2715 assert_eq!(build_result.changes.len(), 0);
2716 }
2717
2718 #[test]
2719 fn test_add_partition_statistics() {
2720 let builder = builder_without_changes(FormatVersion::V2);
2721
2722 let statistics = PartitionStatisticsFile {
2723 snapshot_id: 3055729675574597004,
2724 statistics_path: "s3://a/b/partition-stats.parquet".to_string(),
2725 file_size_in_bytes: 43,
2726 };
2727
2728 let build_result = builder
2729 .set_partition_statistics(statistics.clone())
2730 .build()
2731 .unwrap();
2732 assert_eq!(
2733 build_result.metadata.partition_statistics,
2734 HashMap::from_iter(vec![(3055729675574597004, statistics.clone())])
2735 );
2736 assert_eq!(build_result.changes, vec![
2737 TableUpdate::SetPartitionStatistics {
2738 partition_statistics: statistics.clone()
2739 }
2740 ]);
2741
2742 let builder = build_result.metadata.into_builder(None);
2744 let build_result = builder
2745 .remove_partition_statistics(statistics.snapshot_id)
2746 .build()
2747 .unwrap();
2748 assert_eq!(build_result.metadata.partition_statistics.len(), 0);
2749 assert_eq!(build_result.changes, vec![
2750 TableUpdate::RemovePartitionStatistics {
2751 snapshot_id: statistics.snapshot_id
2752 }
2753 ]);
2754
2755 let builder = build_result.metadata.into_builder(None);
2757 let build_result = builder
2758 .remove_partition_statistics(statistics.snapshot_id)
2759 .build()
2760 .unwrap();
2761 assert_eq!(build_result.metadata.partition_statistics.len(), 0);
2762 assert_eq!(build_result.changes.len(), 0);
2763 }
2764
2765 #[test]
2766 fn last_update_increased_for_property_only_update() {
2767 let builder = builder_without_changes(FormatVersion::V2);
2768
2769 let metadata = builder.build().unwrap().metadata;
2770 let last_updated_ms = metadata.last_updated_ms;
2771 sleep(std::time::Duration::from_millis(2));
2772
2773 let build_result = metadata
2774 .into_builder(Some(
2775 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
2776 ))
2777 .set_properties(HashMap::from_iter(vec![(
2778 "foo".to_string(),
2779 "bar".to_string(),
2780 )]))
2781 .unwrap()
2782 .build()
2783 .unwrap();
2784
2785 assert!(
2786 build_result.metadata.last_updated_ms > last_updated_ms,
2787 "{} > {}",
2788 build_result.metadata.last_updated_ms,
2789 last_updated_ms
2790 );
2791 }
2792
2793 #[test]
2794 fn test_construct_default_main_branch() {
2795 let file = File::open(format!(
2797 "{}/testdata/table_metadata/{}",
2798 env!("CARGO_MANIFEST_DIR"),
2799 "TableMetadataV2Valid.json"
2800 ))
2801 .unwrap();
2802 let reader = BufReader::new(file);
2803 let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
2804
2805 let table = Table::builder()
2806 .metadata(resp)
2807 .metadata_location("s3://bucket/test/location/metadata/v1.json")
2808 .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
2809 .file_io(FileIO::new_with_memory())
2810 .runtime(test_runtime())
2811 .build()
2812 .unwrap();
2813
2814 assert_eq!(
2815 table.metadata().refs.get(MAIN_BRANCH).unwrap().snapshot_id,
2816 table.metadata().current_snapshot_id().unwrap()
2817 );
2818 }
2819
2820 #[test]
2821 fn test_active_schema_cannot_be_removed() {
2822 let builder = builder_without_changes(FormatVersion::V2);
2823 builder.remove_schemas(&[0]).unwrap_err();
2824 }
2825
2826 #[test]
2827 fn test_remove_schemas() {
2828 let file = File::open(format!(
2829 "{}/testdata/table_metadata/{}",
2830 env!("CARGO_MANIFEST_DIR"),
2831 "TableMetadataV2Valid.json"
2832 ))
2833 .unwrap();
2834 let reader = BufReader::new(file);
2835 let resp = serde_json::from_reader::<_, TableMetadata>(reader).unwrap();
2836
2837 let table = Table::builder()
2838 .metadata(resp)
2839 .metadata_location("s3://bucket/test/location/metadata/v1.json")
2840 .identifier(TableIdent::from_strs(["ns1", "test1"]).unwrap())
2841 .file_io(FileIO::new_with_memory())
2842 .runtime(test_runtime())
2843 .build()
2844 .unwrap();
2845
2846 assert_eq!(2, table.metadata().schemas.len());
2847
2848 {
2849 let meta_data_builder = table.metadata().clone().into_builder(None);
2851 meta_data_builder.remove_schemas(&[1]).unwrap_err();
2852 }
2853
2854 let mut meta_data_builder = table.metadata().clone().into_builder(None);
2855 meta_data_builder = meta_data_builder.remove_schemas(&[0]).unwrap();
2856 let build_result = meta_data_builder.build().unwrap();
2857 assert_eq!(1, build_result.metadata.schemas.len());
2858 assert_eq!(1, build_result.metadata.current_schema_id);
2859 assert_eq!(1, build_result.metadata.current_schema().schema_id());
2860 assert_eq!(1, build_result.changes.len());
2861
2862 let remove_schema_ids =
2863 if let TableUpdate::RemoveSchemas { schema_ids } = &build_result.changes[0] {
2864 schema_ids
2865 } else {
2866 unreachable!("Expected RemoveSchema change")
2867 };
2868 assert_eq!(remove_schema_ids, &[0]);
2869 }
2870
2871 #[test]
2872 fn test_schema_evolution_now_correctly_validates_partition_field_name_conflicts() {
2873 let initial_schema = Schema::builder()
2874 .with_fields(vec![
2875 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2876 ])
2877 .build()
2878 .unwrap();
2879
2880 let partition_spec_with_bucket = UnboundPartitionSpec::builder()
2881 .with_spec_id(0)
2882 .add_partition_field(
2883 UnboundPartitionField::builder()
2884 .source_ids(vec![1])
2885 .name("bucket_data")
2886 .transform(Transform::Bucket(16))
2887 .build()
2888 .unwrap(),
2889 )
2890 .unwrap()
2891 .build();
2892
2893 let metadata = TableMetadataBuilder::new(
2894 initial_schema,
2895 partition_spec_with_bucket,
2896 SortOrder::unsorted_order(),
2897 TEST_LOCATION.to_string(),
2898 FormatVersion::V2,
2899 HashMap::new(),
2900 )
2901 .unwrap()
2902 .build()
2903 .unwrap()
2904 .metadata;
2905
2906 let partition_field_names: Vec<String> = metadata
2907 .default_partition_spec()
2908 .fields()
2909 .iter()
2910 .map(|f| f.name.clone())
2911 .collect();
2912 assert!(partition_field_names.contains(&"bucket_data".to_string()));
2913
2914 let evolved_schema = Schema::builder()
2915 .with_fields(vec![
2916 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2917 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
2919 ])
2920 .build()
2921 .unwrap();
2922
2923 let builder = metadata.into_builder(Some(
2924 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
2925 ));
2926
2927 let result = builder.add_current_schema(evolved_schema);
2929
2930 assert!(result.is_err());
2931 let error = result.unwrap_err();
2932 let error_message = error.message();
2933 assert!(error_message.contains("Cannot set current schema"));
2934 assert!(error_message.contains("sourced from field id 1"));
2935 }
2936
2937 #[test]
2938 fn test_schema_evolution_should_validate_on_schema_add_not_metadata_build() {
2939 let initial_schema = Schema::builder()
2940 .with_fields(vec![
2941 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2942 ])
2943 .build()
2944 .unwrap();
2945
2946 let partition_spec = UnboundPartitionSpec::builder()
2947 .with_spec_id(0)
2948 .add_partition_field(
2949 UnboundPartitionField::builder()
2950 .source_ids(vec![1])
2951 .name("partition_col")
2952 .transform(Transform::Bucket(16))
2953 .build()
2954 .unwrap(),
2955 )
2956 .unwrap()
2957 .build();
2958
2959 let metadata = TableMetadataBuilder::new(
2960 initial_schema,
2961 partition_spec,
2962 SortOrder::unsorted_order(),
2963 TEST_LOCATION.to_string(),
2964 FormatVersion::V2,
2965 HashMap::new(),
2966 )
2967 .unwrap()
2968 .build()
2969 .unwrap()
2970 .metadata;
2971
2972 let non_conflicting_schema = Schema::builder()
2973 .with_fields(vec![
2974 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2975 NestedField::required(2, "new_field", Type::Primitive(PrimitiveType::Int)).into(),
2976 ])
2977 .build()
2978 .unwrap();
2979
2980 let result = metadata
2982 .clone()
2983 .into_builder(Some("test_location".to_string()))
2984 .add_current_schema(non_conflicting_schema)
2985 .unwrap()
2986 .build();
2987
2988 assert!(result.is_ok());
2989 }
2990
2991 #[test]
2992 fn test_partition_spec_evolution_validates_schema_field_name_conflicts() {
2993 let initial_schema = Schema::builder()
2994 .with_fields(vec![
2995 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
2996 NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
2997 .into(),
2998 ])
2999 .build()
3000 .unwrap();
3001
3002 let partition_spec = UnboundPartitionSpec::builder()
3003 .with_spec_id(0)
3004 .add_partition_field(
3005 UnboundPartitionField::builder()
3006 .source_ids(vec![1])
3007 .name("data_bucket")
3008 .transform(Transform::Bucket(16))
3009 .build()
3010 .unwrap(),
3011 )
3012 .unwrap()
3013 .build();
3014
3015 let metadata = TableMetadataBuilder::new(
3016 initial_schema,
3017 partition_spec,
3018 SortOrder::unsorted_order(),
3019 TEST_LOCATION.to_string(),
3020 FormatVersion::V2,
3021 HashMap::new(),
3022 )
3023 .unwrap()
3024 .build()
3025 .unwrap()
3026 .metadata;
3027
3028 let builder = metadata.into_builder(Some(
3029 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3030 ));
3031
3032 let conflicting_partition_spec = UnboundPartitionSpec::builder()
3033 .with_spec_id(1)
3034 .add_partition_field(
3035 UnboundPartitionField::builder()
3036 .source_ids(vec![1])
3037 .name("existing_field")
3038 .transform(Transform::Bucket(8))
3039 .build()
3040 .unwrap(),
3041 )
3042 .unwrap()
3043 .build();
3044
3045 let result = builder.add_partition_spec(conflicting_partition_spec);
3046
3047 assert!(result.is_err());
3048 let error = result.unwrap_err();
3049 let error_message = error.message();
3050 assert!(error_message.contains(
3052 "Cannot create partition with name 'existing_field' that conflicts with schema field"
3053 ));
3054 assert!(error_message.contains("and is not an identity or void transform"));
3055 }
3056
3057 #[test]
3058 fn test_schema_evolution_validates_against_all_historical_schemas() {
3059 let initial_schema = Schema::builder()
3061 .with_fields(vec![
3062 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3063 NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
3064 .into(),
3065 ])
3066 .build()
3067 .unwrap();
3068
3069 let partition_spec = UnboundPartitionSpec::builder()
3070 .with_spec_id(0)
3071 .add_partition_field(
3072 UnboundPartitionField::builder()
3073 .source_ids(vec![1])
3074 .name("bucket_data")
3075 .transform(Transform::Bucket(16))
3076 .build()
3077 .unwrap(),
3078 )
3079 .unwrap()
3080 .build();
3081
3082 let metadata = TableMetadataBuilder::new(
3083 initial_schema,
3084 partition_spec,
3085 SortOrder::unsorted_order(),
3086 TEST_LOCATION.to_string(),
3087 FormatVersion::V2,
3088 HashMap::new(),
3089 )
3090 .unwrap()
3091 .build()
3092 .unwrap()
3093 .metadata;
3094
3095 let second_schema = Schema::builder()
3097 .with_fields(vec![
3098 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3099 NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3100 .into(),
3101 ])
3102 .build()
3103 .unwrap();
3104
3105 let metadata = metadata
3106 .into_builder(Some("test_location".to_string()))
3107 .add_current_schema(second_schema)
3108 .unwrap()
3109 .build()
3110 .unwrap()
3111 .metadata;
3112
3113 let third_schema = Schema::builder()
3117 .with_fields(vec![
3118 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3119 NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3120 .into(),
3121 NestedField::required(4, "existing_field", Type::Primitive(PrimitiveType::Int))
3122 .into(),
3123 ])
3124 .build()
3125 .unwrap();
3126
3127 let builder = metadata
3128 .clone()
3129 .into_builder(Some("test_location".to_string()));
3130
3131 let result = builder.add_current_schema(third_schema);
3133 assert!(result.is_ok());
3134
3135 let conflicting_schema = Schema::builder()
3138 .with_fields(vec![
3139 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3140 NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3141 .into(),
3142 NestedField::required(4, "existing_field", Type::Primitive(PrimitiveType::Int))
3143 .into(),
3144 NestedField::required(5, "bucket_data", Type::Primitive(PrimitiveType::String))
3145 .into(), ])
3147 .build()
3148 .unwrap();
3149
3150 let builder2 = metadata.into_builder(Some("test_location".to_string()));
3151 let result2 = builder2.add_current_schema(conflicting_schema);
3152
3153 assert!(result2.is_err());
3156 let error = result2.unwrap_err();
3157 assert!(error.message().contains("Cannot set current schema"));
3158 assert!(error.message().contains("sourced from field id 1"));
3159 }
3160
3161 #[test]
3162 fn test_schema_evolution_allows_existing_partition_field_if_exists_in_historical_schema() {
3163 let initial_schema = Schema::builder()
3165 .with_fields(vec![
3166 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3167 NestedField::required(2, "partition_data", Type::Primitive(PrimitiveType::Int))
3168 .into(),
3169 ])
3170 .build()
3171 .unwrap();
3172
3173 let partition_spec = UnboundPartitionSpec::builder()
3174 .with_spec_id(0)
3175 .add_partition_field(
3176 UnboundPartitionField::builder()
3177 .source_ids(vec![2])
3178 .name("partition_data")
3179 .transform(Transform::Identity)
3180 .build()
3181 .unwrap(),
3182 )
3183 .unwrap()
3184 .build();
3185
3186 let metadata = TableMetadataBuilder::new(
3187 initial_schema,
3188 partition_spec,
3189 SortOrder::unsorted_order(),
3190 TEST_LOCATION.to_string(),
3191 FormatVersion::V2,
3192 HashMap::new(),
3193 )
3194 .unwrap()
3195 .build()
3196 .unwrap()
3197 .metadata;
3198
3199 let evolved_schema = Schema::builder()
3201 .with_fields(vec![
3202 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3203 NestedField::required(2, "partition_data", Type::Primitive(PrimitiveType::Int))
3204 .into(),
3205 NestedField::required(3, "new_field", Type::Primitive(PrimitiveType::String))
3206 .into(),
3207 ])
3208 .build()
3209 .unwrap();
3210
3211 let result = metadata
3213 .into_builder(Some("test_location".to_string()))
3214 .add_current_schema(evolved_schema);
3215
3216 assert!(result.is_ok());
3217 }
3218
3219 fn bucket_partitioned_metadata() -> TableMetadata {
3221 let schema = Schema::builder()
3222 .with_fields(vec![
3223 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3224 ])
3225 .build()
3226 .unwrap();
3227 let spec = UnboundPartitionSpec::builder()
3228 .with_spec_id(0)
3229 .add_partition_field(
3230 UnboundPartitionField::builder()
3231 .source_ids(vec![1])
3232 .name("bucket_data")
3233 .transform(Transform::Bucket(16))
3234 .build()
3235 .unwrap(),
3236 )
3237 .unwrap()
3238 .build();
3239
3240 TableMetadataBuilder::new(
3241 schema,
3242 spec,
3243 SortOrder::unsorted_order(),
3244 TEST_LOCATION.to_string(),
3245 FormatVersion::V2,
3246 HashMap::new(),
3247 )
3248 .unwrap()
3249 .build()
3250 .unwrap()
3251 .metadata
3252 }
3253
3254 fn unpartition(metadata: TableMetadata, file: &str) -> TableMetadata {
3256 let metadata = metadata
3257 .into_builder(Some(file.to_string()))
3258 .add_default_partition_spec(UnboundPartitionSpec::builder().build())
3259 .unwrap()
3260 .build()
3261 .unwrap()
3262 .metadata;
3263 assert_ne!(metadata.default_partition_spec().spec_id(), 0);
3264 assert!(metadata.partition_specs.contains_key(&0));
3265 metadata
3266 }
3267
3268 #[test]
3269 fn test_schema_evolution_allows_a_superseded_partition_field_name_once_its_source_is_dropped() {
3270 let schema = Schema::builder()
3273 .with_fields(vec![
3274 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3275 ])
3276 .build()
3277 .unwrap();
3278
3279 let metadata = unpartition(
3280 bucket_partitioned_metadata(),
3281 "s3://bucket/t/metadata/m1.json",
3282 )
3283 .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3284 .add_current_schema(schema.clone())
3285 .unwrap()
3286 .build()
3287 .unwrap()
3288 .metadata;
3289
3290 assert_eq!(metadata.current_schema().as_struct(), schema.as_struct());
3291 assert!(metadata.current_schema().field_by_name("data").is_none());
3292 assert_eq!(
3294 metadata.partition_spec_by_id(0).unwrap().fields()[0].name,
3295 "bucket_data"
3296 );
3297 }
3298
3299 #[test]
3300 fn test_schema_evolution_rejects_a_partition_field_name_while_its_source_remains() {
3301 let schema = Schema::builder()
3304 .with_fields(vec![
3305 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3306 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3307 ])
3308 .build()
3309 .unwrap();
3310
3311 let error = bucket_partitioned_metadata()
3312 .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3313 .add_current_schema(schema.clone())
3314 .unwrap_err();
3315 assert_eq!(error.kind(), ErrorKind::DataInvalid);
3316 assert!(
3317 error.message().contains("sourced from field id 1"),
3318 "{error}"
3319 );
3320
3321 let error = unpartition(
3322 bucket_partitioned_metadata(),
3323 "s3://bucket/t/metadata/m1.json",
3324 )
3325 .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3326 .add_current_schema(schema)
3327 .unwrap_err();
3328 assert_eq!(error.kind(), ErrorKind::DataInvalid);
3329 assert!(
3330 error.message().contains("sourced from field id 1"),
3331 "{error}"
3332 );
3333 }
3334
3335 #[test]
3336 fn test_schema_evolution_partition_name_check_is_order_independent() {
3337 let schema = Schema::builder()
3340 .with_fields(vec![
3341 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3342 ])
3343 .build()
3344 .unwrap();
3345 let unpartitioned = UnboundPartitionSpec::builder().build();
3346
3347 let schema_first = bucket_partitioned_metadata()
3348 .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3349 .add_current_schema(schema.clone())
3350 .and_then(|b| b.add_default_partition_spec(unpartitioned.clone()))
3351 .and_then(|b| b.build());
3352 let spec_first = bucket_partitioned_metadata()
3353 .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3354 .add_default_partition_spec(unpartitioned)
3355 .and_then(|b| b.add_current_schema(schema))
3356 .and_then(|b| b.build());
3357
3358 assert!(schema_first.is_ok(), "{:?}", schema_first.err());
3359 assert!(spec_first.is_ok(), "{:?}", spec_first.err());
3360 }
3361
3362 #[test]
3363 fn test_set_current_schema_rejects_a_partition_field_name_from_a_spec_added_after_the_schema() {
3364 let schema = Schema::builder()
3365 .with_fields(vec![
3366 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3367 ])
3368 .build()
3369 .unwrap();
3370 let conflicting_schema = Schema::builder()
3371 .with_fields(vec![
3372 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3373 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3374 ])
3375 .build()
3376 .unwrap();
3377 let spec = UnboundPartitionSpec::builder()
3378 .add_partition_field(
3379 UnboundPartitionField::builder()
3380 .source_ids(vec![1])
3381 .name("bucket_data")
3382 .transform(Transform::Bucket(16))
3383 .build()
3384 .unwrap(),
3385 )
3386 .unwrap()
3387 .build();
3388
3389 let error = TableMetadataBuilder::new(
3390 schema,
3391 UnboundPartitionSpec::builder().build(),
3392 SortOrder::unsorted_order(),
3393 TEST_LOCATION.to_string(),
3394 FormatVersion::V2,
3395 HashMap::new(),
3396 )
3397 .unwrap()
3398 .add_schema(conflicting_schema)
3399 .and_then(|b| b.add_partition_spec(spec))
3400 .and_then(|b| b.set_current_schema(TableMetadataBuilder::LAST_ADDED))
3401 .unwrap_err();
3402 assert_eq!(error.kind(), ErrorKind::DataInvalid);
3403 assert!(
3404 error.message().contains("sourced from field id 1"),
3405 "{error}"
3406 );
3407 }
3408
3409 #[test]
3410 fn test_schema_with_a_conflicting_partition_field_name_can_be_added_but_not_made_current() {
3411 let conflicting_schema = Schema::builder()
3412 .with_fields(vec![
3413 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3414 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3415 ])
3416 .build()
3417 .unwrap();
3418
3419 let metadata = bucket_partitioned_metadata()
3420 .into_builder(Some("s3://bucket/t/metadata/m1.json".to_string()))
3421 .add_schema(conflicting_schema.clone())
3422 .unwrap()
3423 .build()
3424 .unwrap()
3425 .metadata;
3426 assert_eq!(metadata.current_schema_id(), 0);
3427 assert_eq!(
3428 metadata.schema_by_id(1).unwrap().as_struct(),
3429 conflicting_schema.as_struct()
3430 );
3431
3432 let error = metadata
3433 .into_builder(Some("s3://bucket/t/metadata/m2.json".to_string()))
3434 .set_current_schema(1)
3435 .unwrap_err();
3436 assert_eq!(error.kind(), ErrorKind::DataInvalid);
3437 assert!(
3438 error.message().contains("sourced from field id 1"),
3439 "{error}"
3440 );
3441 }
3442
3443 #[test]
3444 fn test_schema_evolution_prevents_new_field_conflicting_with_partition_field() {
3445 let initial_schema = Schema::builder()
3447 .with_fields(vec![
3448 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3449 ])
3450 .build()
3451 .unwrap();
3452
3453 let partition_spec = UnboundPartitionSpec::builder()
3454 .with_spec_id(0)
3455 .add_partition_field(
3456 UnboundPartitionField::builder()
3457 .source_ids(vec![1])
3458 .name("bucket_data")
3459 .transform(Transform::Bucket(16))
3460 .build()
3461 .unwrap(),
3462 )
3463 .unwrap()
3464 .build();
3465
3466 let metadata = TableMetadataBuilder::new(
3467 initial_schema,
3468 partition_spec,
3469 SortOrder::unsorted_order(),
3470 TEST_LOCATION.to_string(),
3471 FormatVersion::V2,
3472 HashMap::new(),
3473 )
3474 .unwrap()
3475 .build()
3476 .unwrap()
3477 .metadata;
3478
3479 let conflicting_schema = Schema::builder()
3481 .with_fields(vec![
3482 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3483 NestedField::required(2, "bucket_data", Type::Primitive(PrimitiveType::Int)).into(),
3485 ])
3486 .build()
3487 .unwrap();
3488
3489 let builder = metadata.into_builder(Some("test_location".to_string()));
3490 let result = builder.add_current_schema(conflicting_schema);
3491
3492 assert!(result.is_err());
3495 let error = result.unwrap_err();
3496 assert!(error.message().contains("Cannot set current schema"));
3497 assert!(error.message().contains("sourced from field id 1"));
3498 }
3499
3500 #[test]
3501 fn test_partition_spec_evolution_allows_non_conflicting_names() {
3502 let initial_schema = Schema::builder()
3503 .with_fields(vec![
3504 NestedField::required(1, "data", Type::Primitive(PrimitiveType::String)).into(),
3505 NestedField::required(2, "existing_field", Type::Primitive(PrimitiveType::Int))
3506 .into(),
3507 ])
3508 .build()
3509 .unwrap();
3510
3511 let partition_spec = UnboundPartitionSpec::builder()
3512 .with_spec_id(0)
3513 .add_partition_field(
3514 UnboundPartitionField::builder()
3515 .source_ids(vec![1])
3516 .name("data_bucket")
3517 .transform(Transform::Bucket(16))
3518 .build()
3519 .unwrap(),
3520 )
3521 .unwrap()
3522 .build();
3523
3524 let metadata = TableMetadataBuilder::new(
3525 initial_schema,
3526 partition_spec,
3527 SortOrder::unsorted_order(),
3528 TEST_LOCATION.to_string(),
3529 FormatVersion::V2,
3530 HashMap::new(),
3531 )
3532 .unwrap()
3533 .build()
3534 .unwrap()
3535 .metadata;
3536
3537 let builder = metadata.into_builder(Some(
3538 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3539 ));
3540
3541 let non_conflicting_partition_spec = UnboundPartitionSpec::builder()
3543 .with_spec_id(1)
3544 .add_partition_field(
3545 UnboundPartitionField::builder()
3546 .source_ids(vec![2])
3547 .name("new_partition_field")
3548 .transform(Transform::Bucket(8))
3549 .build()
3550 .unwrap(),
3551 )
3552 .unwrap()
3553 .build();
3554
3555 let result = builder.add_partition_spec(non_conflicting_partition_spec);
3556
3557 assert!(result.is_ok());
3558 }
3559
3560 #[test]
3561 fn test_row_lineage_addition() {
3562 let new_rows = 30;
3563 let base = builder_without_changes(FormatVersion::V3)
3564 .build()
3565 .unwrap()
3566 .metadata;
3567 let add_rows = Snapshot::builder()
3568 .with_snapshot_id(0)
3569 .with_timestamp_ms(base.last_updated_ms + 1)
3570 .with_sequence_number(0)
3571 .with_schema_id(0)
3572 .with_manifest_list("foo")
3573 .with_parent_snapshot_id(None)
3574 .with_summary(Summary {
3575 operation: Operation::Append,
3576 additional_properties: HashMap::new(),
3577 })
3578 .with_row_range(base.next_row_id(), new_rows)
3579 .build();
3580
3581 let first_addition = base
3582 .into_builder(None)
3583 .add_snapshot(add_rows.clone())
3584 .unwrap()
3585 .build()
3586 .unwrap()
3587 .metadata;
3588
3589 assert_eq!(first_addition.next_row_id(), new_rows);
3590
3591 let add_more_rows = Snapshot::builder()
3592 .with_snapshot_id(1)
3593 .with_timestamp_ms(first_addition.last_updated_ms + 1)
3594 .with_sequence_number(1)
3595 .with_schema_id(0)
3596 .with_manifest_list("foo")
3597 .with_parent_snapshot_id(Some(0))
3598 .with_summary(Summary {
3599 operation: Operation::Append,
3600 additional_properties: HashMap::new(),
3601 })
3602 .with_row_range(first_addition.next_row_id(), new_rows)
3603 .build();
3604
3605 let second_addition = first_addition
3606 .into_builder(None)
3607 .add_snapshot(add_more_rows)
3608 .unwrap()
3609 .build()
3610 .unwrap()
3611 .metadata;
3612 assert_eq!(second_addition.next_row_id(), new_rows * 2);
3613 }
3614
3615 #[test]
3616 fn test_row_lineage_invalid_snapshot() {
3617 let new_rows = 30;
3618 let base = builder_without_changes(FormatVersion::V3)
3619 .build()
3620 .unwrap()
3621 .metadata;
3622
3623 let add_rows = Snapshot::builder()
3625 .with_snapshot_id(0)
3626 .with_timestamp_ms(base.last_updated_ms + 1)
3627 .with_sequence_number(0)
3628 .with_schema_id(0)
3629 .with_manifest_list("foo")
3630 .with_parent_snapshot_id(None)
3631 .with_summary(Summary {
3632 operation: Operation::Append,
3633 additional_properties: HashMap::new(),
3634 })
3635 .with_row_range(base.next_row_id(), new_rows)
3636 .build();
3637
3638 let added = base
3639 .into_builder(None)
3640 .add_snapshot(add_rows)
3641 .unwrap()
3642 .build()
3643 .unwrap()
3644 .metadata;
3645
3646 let invalid_new_rows = Snapshot::builder()
3647 .with_snapshot_id(1)
3648 .with_timestamp_ms(added.last_updated_ms + 1)
3649 .with_sequence_number(1)
3650 .with_schema_id(0)
3651 .with_manifest_list("foo")
3652 .with_parent_snapshot_id(Some(0))
3653 .with_summary(Summary {
3654 operation: Operation::Append,
3655 additional_properties: HashMap::new(),
3656 })
3657 .with_row_range(added.next_row_id() - 1, 10)
3659 .build();
3660
3661 let err = added
3662 .into_builder(None)
3663 .add_snapshot(invalid_new_rows)
3664 .unwrap_err();
3665 assert!(
3666 err.to_string().contains(
3667 "Cannot add a snapshot, first-row-id is behind table next-row-id: 29 < 30"
3668 )
3669 );
3670 }
3671
3672 #[test]
3673 fn test_row_lineage_append_branch() {
3674 let branch = "some_branch";
3678
3679 let base = builder_without_changes(FormatVersion::V3)
3681 .build()
3682 .unwrap()
3683 .metadata;
3684
3685 assert_eq!(base.next_row_id(), 0);
3687
3688 let branch_snapshot_1 = Snapshot::builder()
3690 .with_snapshot_id(1)
3691 .with_timestamp_ms(base.last_updated_ms + 1)
3692 .with_sequence_number(0)
3693 .with_schema_id(0)
3694 .with_manifest_list("foo")
3695 .with_parent_snapshot_id(None)
3696 .with_summary(Summary {
3697 operation: Operation::Append,
3698 additional_properties: HashMap::new(),
3699 })
3700 .with_row_range(base.next_row_id(), 30)
3701 .build();
3702
3703 let table_after_branch_1 = base
3704 .into_builder(None)
3705 .set_branch_snapshot(branch_snapshot_1.clone(), branch)
3706 .unwrap()
3707 .build()
3708 .unwrap()
3709 .metadata;
3710
3711 assert!(table_after_branch_1.current_snapshot().is_none());
3713
3714 let branch_ref = table_after_branch_1.refs.get(branch).unwrap();
3716 let branch_snap_1 = table_after_branch_1
3717 .snapshots
3718 .get(&branch_ref.snapshot_id)
3719 .unwrap();
3720 assert_eq!(branch_snap_1.first_row_id(), Some(0));
3721
3722 assert_eq!(table_after_branch_1.next_row_id(), 30);
3724
3725 let main_snapshot = Snapshot::builder()
3727 .with_snapshot_id(2)
3728 .with_timestamp_ms(table_after_branch_1.last_updated_ms + 1)
3729 .with_sequence_number(1)
3730 .with_schema_id(0)
3731 .with_manifest_list("bar")
3732 .with_parent_snapshot_id(None)
3733 .with_summary(Summary {
3734 operation: Operation::Append,
3735 additional_properties: HashMap::new(),
3736 })
3737 .with_row_range(table_after_branch_1.next_row_id(), 28)
3738 .build();
3739
3740 let table_after_main = table_after_branch_1
3741 .into_builder(None)
3742 .add_snapshot(main_snapshot.clone())
3743 .unwrap()
3744 .set_ref(MAIN_BRANCH, SnapshotReference {
3745 snapshot_id: main_snapshot.snapshot_id(),
3746 retention: SnapshotRetention::Branch {
3747 min_snapshots_to_keep: None,
3748 max_snapshot_age_ms: None,
3749 max_ref_age_ms: None,
3750 },
3751 })
3752 .unwrap()
3753 .build()
3754 .unwrap()
3755 .metadata;
3756
3757 let current_snapshot = table_after_main.current_snapshot().unwrap();
3759 assert_eq!(current_snapshot.first_row_id(), Some(30));
3760
3761 assert_eq!(table_after_main.next_row_id(), 58);
3763
3764 let branch_snapshot_2 = Snapshot::builder()
3766 .with_snapshot_id(3)
3767 .with_timestamp_ms(table_after_main.last_updated_ms + 1)
3768 .with_sequence_number(2)
3769 .with_schema_id(0)
3770 .with_manifest_list("baz")
3771 .with_parent_snapshot_id(Some(branch_snapshot_1.snapshot_id()))
3772 .with_summary(Summary {
3773 operation: Operation::Append,
3774 additional_properties: HashMap::new(),
3775 })
3776 .with_row_range(table_after_main.next_row_id(), 21)
3777 .build();
3778
3779 let table_after_branch_2 = table_after_main
3780 .into_builder(None)
3781 .set_branch_snapshot(branch_snapshot_2.clone(), branch)
3782 .unwrap()
3783 .build()
3784 .unwrap()
3785 .metadata;
3786
3787 let branch_ref_2 = table_after_branch_2.refs.get(branch).unwrap();
3789 let branch_snap_2 = table_after_branch_2
3790 .snapshots
3791 .get(&branch_ref_2.snapshot_id)
3792 .unwrap();
3793 assert_eq!(branch_snap_2.first_row_id(), Some(58));
3794
3795 assert_eq!(table_after_branch_2.next_row_id(), 79);
3797 }
3798
3799 #[test]
3800 fn test_encryption_keys() {
3801 let builder = builder_without_changes(FormatVersion::V2);
3802
3803 let encryption_key_1 = EncryptedKey::builder()
3805 .key_id("key-1")
3806 .encrypted_key_metadata(vec![1, 2, 3, 4])
3807 .encrypted_by_id("encryption-service-1")
3808 .properties(HashMap::from_iter(vec![(
3809 "algorithm".to_string(),
3810 "AES-256".to_string(),
3811 )]))
3812 .build();
3813
3814 let encryption_key_2 = EncryptedKey::builder()
3815 .key_id("key-2")
3816 .encrypted_key_metadata(vec![5, 6, 7, 8])
3817 .encrypted_by_id("encryption-service-2")
3818 .properties(HashMap::new())
3819 .build();
3820
3821 let build_result = builder
3823 .add_encryption_key(encryption_key_1.clone())
3824 .build()
3825 .unwrap();
3826
3827 assert_eq!(build_result.changes.len(), 1);
3828 assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3829 assert_eq!(
3830 build_result.metadata.encryption_key("key-1"),
3831 Some(&encryption_key_1)
3832 );
3833 assert_eq!(build_result.changes[0], TableUpdate::AddEncryptionKey {
3834 encryption_key: encryption_key_1.clone()
3835 });
3836
3837 let build_result = build_result
3839 .metadata
3840 .into_builder(Some(
3841 "s3://bucket/test/location/metadata/metadata1.json".to_string(),
3842 ))
3843 .add_encryption_key(encryption_key_2.clone())
3844 .build()
3845 .unwrap();
3846
3847 assert_eq!(build_result.changes.len(), 1);
3848 assert_eq!(build_result.metadata.encryption_keys.len(), 2);
3849 assert_eq!(
3850 build_result.metadata.encryption_key("key-1"),
3851 Some(&encryption_key_1)
3852 );
3853 assert_eq!(
3854 build_result.metadata.encryption_key("key-2"),
3855 Some(&encryption_key_2)
3856 );
3857 assert_eq!(build_result.changes[0], TableUpdate::AddEncryptionKey {
3858 encryption_key: encryption_key_2.clone()
3859 });
3860
3861 let build_result = build_result
3863 .metadata
3864 .into_builder(Some(
3865 "s3://bucket/test/location/metadata/metadata2.json".to_string(),
3866 ))
3867 .add_encryption_key(encryption_key_1.clone())
3868 .build()
3869 .unwrap();
3870
3871 assert_eq!(build_result.changes.len(), 0);
3872 assert_eq!(build_result.metadata.encryption_keys.len(), 2);
3873
3874 let build_result = build_result
3876 .metadata
3877 .into_builder(Some(
3878 "s3://bucket/test/location/metadata/metadata3.json".to_string(),
3879 ))
3880 .remove_encryption_key("key-1")
3881 .build()
3882 .unwrap();
3883
3884 assert_eq!(build_result.changes.len(), 1);
3885 assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3886 assert_eq!(build_result.metadata.encryption_key("key-1"), None);
3887 assert_eq!(
3888 build_result.metadata.encryption_key("key-2"),
3889 Some(&encryption_key_2)
3890 );
3891 assert_eq!(build_result.changes[0], TableUpdate::RemoveEncryptionKey {
3892 key_id: "key-1".to_string()
3893 });
3894
3895 let build_result = build_result
3897 .metadata
3898 .into_builder(Some(
3899 "s3://bucket/test/location/metadata/metadata4.json".to_string(),
3900 ))
3901 .remove_encryption_key("non-existent-key")
3902 .build()
3903 .unwrap();
3904
3905 assert_eq!(build_result.changes.len(), 0);
3906 assert_eq!(build_result.metadata.encryption_keys.len(), 1);
3907
3908 let keys = build_result
3910 .metadata
3911 .encryption_keys_iter()
3912 .collect::<Vec<_>>();
3913 assert_eq!(keys.len(), 1);
3914 assert_eq!(keys[0], &encryption_key_2);
3915
3916 let build_result = build_result
3918 .metadata
3919 .into_builder(Some(
3920 "s3://bucket/test/location/metadata/metadata5.json".to_string(),
3921 ))
3922 .remove_encryption_key("key-2")
3923 .build()
3924 .unwrap();
3925
3926 assert_eq!(build_result.changes.len(), 1);
3927 assert_eq!(build_result.metadata.encryption_keys.len(), 0);
3928 assert_eq!(build_result.metadata.encryption_key("key-2"), None);
3929 assert_eq!(build_result.changes[0], TableUpdate::RemoveEncryptionKey {
3930 key_id: "key-2".to_string()
3931 });
3932
3933 let keys = build_result.metadata.encryption_keys_iter();
3935 assert_eq!(keys.len(), 0);
3936 }
3937
3938 #[test]
3939 fn test_partition_field_id_reuse_across_specs() {
3940 let schema = Schema::builder()
3941 .with_fields(vec![
3942 NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
3943 NestedField::required(2, "data", Type::Primitive(PrimitiveType::String)).into(),
3944 NestedField::required(3, "timestamp", Type::Primitive(PrimitiveType::Timestamp))
3945 .into(),
3946 ])
3947 .build()
3948 .unwrap();
3949
3950 let initial_spec = UnboundPartitionSpec::builder()
3952 .add_partition_field(
3953 UnboundPartitionField::builder()
3954 .source_ids(vec![1])
3955 .name("id")
3956 .transform(Transform::Identity)
3957 .build()
3958 .unwrap(),
3959 )
3960 .unwrap()
3961 .build();
3962
3963 let mut metadata = TableMetadataBuilder::new(
3964 schema,
3965 initial_spec,
3966 SortOrder::unsorted_order(),
3967 "s3://bucket/table".to_string(),
3968 FormatVersion::V2,
3969 HashMap::new(),
3970 )
3971 .unwrap()
3972 .build()
3973 .unwrap()
3974 .metadata;
3975
3976 let spec1 = UnboundPartitionSpec::builder()
3978 .add_partition_field(
3979 UnboundPartitionField::builder()
3980 .source_ids(vec![2])
3981 .name("data_bucket")
3982 .transform(Transform::Bucket(10))
3983 .build()
3984 .unwrap(),
3985 )
3986 .unwrap()
3987 .build();
3988 let builder = metadata.into_builder(Some("s3://bucket/table/metadata/v1.json".to_string()));
3989 let result = builder.add_partition_spec(spec1).unwrap().build().unwrap();
3990 metadata = result.metadata;
3991
3992 let spec2 = UnboundPartitionSpec::builder()
3995 .add_partition_field(
3996 UnboundPartitionField::builder()
3997 .source_ids(vec![1])
3998 .name("id")
3999 .transform(Transform::Identity)
4000 .build()
4001 .unwrap(),
4002 ) .unwrap()
4004 .add_partition_field(
4005 UnboundPartitionField::builder()
4006 .source_ids(vec![2])
4007 .name("data_bucket")
4008 .transform(Transform::Bucket(10))
4009 .build()
4010 .unwrap(),
4011 ) .unwrap()
4013 .add_partition_field(
4014 UnboundPartitionField::builder()
4015 .source_ids(vec![3])
4016 .name("year")
4017 .transform(Transform::Year)
4018 .build()
4019 .unwrap(),
4020 ) .unwrap()
4022 .build();
4023 let builder = metadata.into_builder(Some("s3://bucket/table/metadata/v2.json".to_string()));
4024 let result = builder.add_partition_spec(spec2).unwrap().build().unwrap();
4025
4026 let spec2 = result.metadata.partition_spec_by_id(2).unwrap();
4028 let field_ids: Vec<i32> = spec2.fields().iter().map(|f| f.field_id).collect();
4029 assert_eq!(field_ids, vec![1000, 1001, 1002]); }
4031
4032 #[test]
4033 fn test_highest_equivalent_field_ids_is_independent_of_field_order() {
4034 let field = |source_id, field_id, transform| PartitionField {
4035 source_id,
4036 field_id,
4037 name: format!("f{field_id}"),
4038 transform,
4039 };
4040 let fields = [
4041 field(1, 1000, Transform::Identity),
4042 field(2, 1003, Transform::Identity),
4043 field(1, 1006, Transform::Identity),
4044 field(1, 1005, Transform::Bucket(16)),
4045 field(2, 1001, Transform::Identity),
4046 field(1, 1002, Transform::Identity),
4047 field(1, 1004, Transform::Bucket(16)),
4048 ];
4049 let expected = HashMap::from([
4050 ((vec![1], Transform::Identity), 1006),
4051 ((vec![2], Transform::Identity), 1003),
4052 ((vec![1], Transform::Bucket(16)), 1005),
4053 ]);
4054
4055 assert_eq!(
4056 TableMetadataBuilder::highest_equivalent_field_ids(fields.iter()),
4057 expected
4058 );
4059 assert_eq!(
4060 TableMetadataBuilder::highest_equivalent_field_ids(fields.iter().rev()),
4061 expected
4062 );
4063 }
4064
4065 #[test]
4066 fn test_partition_field_id_reuse_picks_the_highest_of_several_candidates() {
4067 let metadata: TableMetadata = serde_json::from_value(serde_json::json!({
4071 "format-version": 2,
4072 "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1",
4073 "location": "s3://bucket/test/location",
4074 "last-sequence-number": 0,
4075 "last-updated-ms": 1602638573590i64,
4076 "last-column-id": 2,
4077 "current-schema-id": 0,
4078 "schemas": [{
4079 "type": "struct",
4080 "schema-id": 0,
4081 "fields": [
4082 {"id": 1, "name": "x", "required": true, "type": "long"},
4083 {"id": 2, "name": "y", "required": true, "type": "long"},
4084 ],
4085 }],
4086 "default-spec-id": 3,
4087 "last-partition-id": 1006,
4088 "partition-specs": [
4089 {"spec-id": 0, "fields": [
4090 {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1000},
4091 ]},
4092 {"spec-id": 1, "fields": [
4093 {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1002},
4094 ]},
4095 {"spec-id": 2, "fields": [
4096 {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1004},
4097 ]},
4098 {"spec-id": 3, "fields": [
4099 {"name": "x", "transform": "identity", "source-id": 1, "field-id": 1006},
4100 ]},
4101 ],
4102 "default-sort-order-id": 0,
4103 "sort-orders": [{"order-id": 0, "fields": []}],
4104 }))
4105 .unwrap();
4106
4107 let new_spec = UnboundPartitionSpec::builder()
4110 .add_partition_field(
4111 UnboundPartitionField::builder()
4112 .source_ids(vec![1])
4113 .name("x")
4114 .transform(Transform::Identity)
4115 .build()
4116 .unwrap(),
4117 )
4118 .unwrap()
4119 .add_partition_field(
4120 UnboundPartitionField::builder()
4121 .source_ids(vec![2])
4122 .name("y")
4123 .transform(Transform::Identity)
4124 .build()
4125 .unwrap(),
4126 )
4127 .unwrap()
4128 .build();
4129
4130 let result = metadata
4131 .into_builder(Some(
4132 "s3://bucket/test/location/metadata/v1.json".to_string(),
4133 ))
4134 .add_partition_spec(new_spec)
4135 .unwrap()
4136 .build()
4137 .unwrap();
4138
4139 let spec = result.metadata.partition_spec_by_id(4).unwrap();
4140 let fields: Vec<(i32, i32)> = spec
4141 .fields()
4142 .iter()
4143 .map(|f| (f.source_id, f.field_id))
4144 .collect();
4145 assert_eq!(fields, vec![(1, 1006), (2, 1007)]);
4146 assert_eq!(result.metadata.last_partition_id, 1007);
4147 }
4148}