diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 78841cf47..a057922a9 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -875,6 +875,7 @@ where let current_metadata_location = catalog.table.metadata_location.clone(); let mut retained = BTreeSet::new(); + let mut current_metadata_for_refs = None; let current_metadata_status = if is_valid_table_metadata_location(&parsed_namespace, &parsed_table, ¤t_metadata_location) { retained.insert(current_metadata_location.clone()); @@ -883,6 +884,7 @@ where match serde_json::from_slice::(¤t_metadata_object.data) { Ok(current_metadata) if current_metadata.is_object() => { retained.extend(metadata_log_locations(¤t_metadata, &parsed_namespace, &parsed_table)); + current_metadata_for_refs = Some(current_metadata); TableMetadataPointerStatus::Valid } Ok(_) | Err(_) => TableMetadataPointerStatus::InvalidJson, @@ -908,6 +910,19 @@ where for metadata_location in metadata_locations.iter().rev().take(retain_recent_metadata_files) { retained.insert(metadata_location.clone()); } + if let Some(current_metadata) = current_metadata_for_refs.as_ref() { + retained.extend( + metadata_locations_for_protected_snapshot_refs( + &self.backend, + table_bucket, + &parsed_namespace, + &parsed_table, + current_metadata, + &metadata_locations, + ) + .await?, + ); + } let orphan_metadata_candidate_locations = metadata_locations .into_iter() @@ -977,6 +992,17 @@ where for metadata_location in metadata_locations.iter().rev().take(retain_recent_metadata_files) { retained.insert(metadata_location.clone()); } + retained.extend( + metadata_locations_for_protected_snapshot_refs( + &self.backend, + table_bucket, + &namespace, + &table, + ¤t_metadata, + &metadata_locations, + ) + .await?, + ); let cleanup_candidate_locations = metadata_locations .into_iter() @@ -1080,6 +1106,17 @@ where let mut protected = metadata_log_locations(¤t_metadata, &namespace, &table); protected.insert(entry.metadata_location.clone()); protected.extend(report.retained_metadata_locations.iter().cloned()); + protected.extend( + metadata_locations_for_protected_snapshot_refs( + &self.backend, + table_bucket, + &namespace, + &table, + ¤t_metadata, + &report.cleanup_candidate_locations, + ) + .await?, + ); let mut cleanup_candidate_locations = BTreeSet::new(); let now = OffsetDateTime::now_utc(); @@ -1645,6 +1682,79 @@ fn metadata_log_locations( locations } +async fn metadata_locations_for_protected_snapshot_refs( + backend: &B, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + current_metadata: &serde_json::Value, + metadata_locations: &[String], +) -> TableCatalogStoreResult> +where + B: TableCatalogObjectBackend, +{ + let protected_snapshot_ids = protected_ref_snapshot_ids(current_metadata); + if protected_snapshot_ids.is_empty() { + return Ok(BTreeSet::new()); + } + + let mut retained = BTreeSet::new(); + for metadata_location in metadata_locations { + if !is_valid_table_metadata_location(namespace, table, metadata_location) { + continue; + } + let Some(metadata_object) = backend.read_object(table_bucket, metadata_location).await? else { + continue; + }; + let Ok(metadata) = serde_json::from_slice::(&metadata_object.data) else { + continue; + }; + if metadata_contains_protected_snapshot_ref(&metadata, &protected_snapshot_ids) { + retained.insert(metadata_location.clone()); + } + } + Ok(retained) +} + +fn protected_ref_snapshot_ids(current_metadata: &serde_json::Value) -> BTreeSet { + let mut snapshot_ids = BTreeSet::new(); + let current_snapshot_id = current_metadata + .get("current-snapshot-id") + .and_then(serde_json::Value::as_i64); + let Some(refs) = current_metadata.get("refs").and_then(serde_json::Value::as_object) else { + return snapshot_ids; + }; + + for reference in refs.values() { + if let Some(snapshot_id) = reference.get("snapshot-id").and_then(serde_json::Value::as_i64) + && Some(snapshot_id) != current_snapshot_id + { + snapshot_ids.insert(snapshot_id); + } + } + snapshot_ids +} + +fn metadata_contains_protected_snapshot_ref(metadata: &serde_json::Value, protected_snapshot_ids: &BTreeSet) -> bool { + let current_snapshot_matches = metadata + .get("current-snapshot-id") + .and_then(serde_json::Value::as_i64) + .is_some_and(|snapshot_id| protected_snapshot_ids.contains(&snapshot_id)); + if current_snapshot_matches { + return true; + } + + let Some(refs) = metadata.get("refs").and_then(serde_json::Value::as_object) else { + return false; + }; + refs.values().any(|reference| { + reference + .get("snapshot-id") + .and_then(serde_json::Value::as_i64) + .is_some_and(|snapshot_id| protected_snapshot_ids.contains(&snapshot_id)) + }) +} + fn metadata_candidate_is_past_safety_window(mod_time: Option, now: OffsetDateTime) -> bool { let Some(mod_time) = mod_time else { return false; @@ -2853,6 +2963,71 @@ mod tests { assert_eq!(report.cleanup_candidate_locations, vec![v1, v3]); } + #[tokio::test] + async fn maintenance_dry_run_keeps_metadata_for_protected_snapshot_refs() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let orphan = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let tagged = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let unreferenced = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00004.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &orphan, b"{}".to_vec()).await; + backend + .seed_object( + bucket, + &tagged, + serde_json::to_vec(&serde_json::json!({ + "current-snapshot-id": 10 + })) + .unwrap(), + ) + .await; + backend + .seed_object( + bucket, + &unreferenced, + serde_json::to_vec(&serde_json::json!({ + "current-snapshot-id": 20 + })) + .unwrap(), + ) + .await; + backend + .seed_object( + bucket, + ¤t, + serde_json::to_vec(&serde_json::json!({ + "current-snapshot-id": 30, + "metadata-log": [], + "refs": { + "main": { + "snapshot-id": 30, + "type": "branch" + }, + "audit": { + "snapshot-id": 10, + "type": "tag" + } + } + })) + .unwrap(), + ) + .await; + + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .unwrap(); + + assert!(report.retained_metadata_locations.contains(&tagged)); + assert_eq!(report.cleanup_candidate_locations, vec![orphan, unreferenced]); + } + #[tokio::test] async fn maintenance_dry_run_keeps_recent_metadata_files_and_ignores_non_metadata_objects() { let backend = TestCatalogObjectBackend::default(); @@ -3103,6 +3278,53 @@ mod tests { assert_eq!(report.orphan_metadata_candidate_locations, vec![uncommitted]); } + #[tokio::test] + async fn orphan_metadata_scan_keeps_metadata_for_protected_snapshot_refs() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let orphan = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let tagged = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let current = default_table_metadata_file_path(&namespace, &table, "00003.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + backend.seed_object(bucket, &orphan, b"{}".to_vec()).await; + backend + .seed_object( + bucket, + &tagged, + serde_json::to_vec(&serde_json::json!({ + "current-snapshot-id": 10 + })) + .unwrap(), + ) + .await; + backend + .seed_object( + bucket, + ¤t, + serde_json::to_vec(&serde_json::json!({ + "current-snapshot-id": 20, + "metadata-log": [], + "refs": { + "audit": { + "snapshot-id": 10, + "type": "tag" + } + } + })) + .unwrap(), + ) + .await; + + let report = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + + assert_eq!(report.current_metadata_status, TableMetadataPointerStatus::Valid); + assert_eq!(report.orphan_metadata_candidate_locations, vec![orphan]); + } + #[tokio::test] async fn object_table_catalog_store_commits_with_token_match_and_writes_log() { let backend = TestCatalogObjectBackend::default();