diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 1d4eea293..e9654f211 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -248,6 +248,29 @@ pub(crate) struct TableMetadataMaintenanceReport { pub cleanup_candidate_locations: Vec, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableCatalogExport { + pub table_bucket: TableBucketEntry, + pub namespace: NamespaceEntry, + pub table: TableEntry, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableMetadataPointerStatus { + Valid, + MissingObject, + InvalidLocation, + InvalidJson, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct TableCatalogDiagnosticsReport { + pub catalog: TableCatalogExport, + pub current_metadata_status: TableMetadataPointerStatus, + pub orphan_metadata_candidate_locations: Vec, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum TableCatalogStoreError { NotFound(String), @@ -515,6 +538,109 @@ where Ok(()) } + pub(crate) async fn export_table_catalog_entry( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + + let Some((table_bucket_entry, _)) = self + .read_entry::(table_bucket, &self.paths.table_bucket_entry_path()) + .await? + else { + return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); + }; + let Some((namespace_entry, _)) = self + .read_entry::(table_bucket, &self.paths.namespace_entry_path(&namespace)) + .await? + else { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {}/{}", + table_bucket, + namespace.public_name() + ))); + }; + let Some((table_entry, _)) = self + .read_entry::(table_bucket, &self.paths.table_entry_path(&namespace, &table)) + .await? + else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + + Ok(TableCatalogExport { + table_bucket: table_bucket_entry, + namespace: namespace_entry, + table: table_entry, + }) + } + + pub(crate) async fn diagnose_table_catalog( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + retain_recent_metadata_files: usize, + ) -> TableCatalogStoreResult { + let parsed_namespace = parse_namespace_for_store(namespace)?; + let parsed_table = parse_table_for_store(table)?; + let catalog = self.export_table_catalog_entry(table_bucket, namespace, table).await?; + let current_metadata_location = catalog.table.metadata_location.clone(); + + let mut retained = BTreeSet::new(); + let current_metadata_status = + if is_valid_table_metadata_location(&parsed_namespace, &parsed_table, ¤t_metadata_location) { + retained.insert(current_metadata_location.clone()); + match self.backend.read_object(table_bucket, ¤t_metadata_location).await? { + Some(current_metadata_object) => { + 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)); + TableMetadataPointerStatus::Valid + } + Ok(_) | Err(_) => TableMetadataPointerStatus::InvalidJson, + } + } + None => TableMetadataPointerStatus::MissingObject, + } + } else { + TableMetadataPointerStatus::InvalidLocation + }; + + let mut metadata_locations = Vec::new(); + let metadata_prefix = format!("{}/", default_table_metadata_dir_path(&parsed_namespace, &parsed_table)); + for object in self.backend.list_objects(table_bucket, &metadata_prefix).await? { + if let Some(metadata_location) = metadata_location_from_metadata_file_path(&parsed_namespace, &parsed_table, &object) + { + metadata_locations.push(metadata_location); + } + } + metadata_locations.sort(); + metadata_locations.dedup(); + + for metadata_location in metadata_locations.iter().rev().take(retain_recent_metadata_files) { + retained.insert(metadata_location.clone()); + } + + let orphan_metadata_candidate_locations = metadata_locations + .into_iter() + .filter(|metadata_location| !retained.contains(metadata_location)) + .collect(); + + Ok(TableCatalogDiagnosticsReport { + catalog, + current_metadata_status, + orphan_metadata_candidate_locations, + }) + } + pub(crate) async fn plan_table_metadata_maintenance( &self, table_bucket: &str, @@ -2368,6 +2494,99 @@ mod tests { assert!(backend.object_exists(bucket, &old).await.unwrap()); } + #[tokio::test] + async fn export_catalog_entry_contains_table_identity_and_pointer() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + + let export = store.export_table_catalog_entry(bucket, "sales", "orders").await.unwrap(); + + assert_eq!(export.table_bucket.table_bucket, bucket); + assert_eq!(export.namespace.namespace, "sales"); + assert_eq!(export.table.namespace, "sales"); + assert_eq!(export.table.table, "orders"); + assert_eq!(export.table.table_id, "table-id"); + assert_eq!(export.table.table_uuid, "table-uuid"); + assert_eq!(export.table.metadata_location, current); + assert_eq!(export.table.version_token, "token-v1"); + assert_eq!(export.table.generation, 1); + } + + #[tokio::test] + async fn consistency_check_reports_missing_metadata_object() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await; + + let report = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + + assert_eq!(report.catalog.table.metadata_location, current.clone()); + assert_eq!(report.current_metadata_status, TableMetadataPointerStatus::MissingObject); + assert!(report.orphan_metadata_candidate_locations.is_empty()); + } + + #[tokio::test] + async fn consistency_check_reports_invalid_metadata_location() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let invalid_metadata = ".rustfs-table/warehouses/default/namespaces/sales/tables/other/metadata/00001.metadata.json"; + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &table, invalid_metadata.to_string())) + .await + .unwrap(); + + let report = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + + assert_eq!(report.catalog.table.metadata_location, invalid_metadata); + assert_eq!(report.current_metadata_status, TableMetadataPointerStatus::InvalidLocation); + assert!(report.orphan_metadata_candidate_locations.is_empty()); + } + + #[tokio::test] + async fn orphan_metadata_scan_does_not_treat_largest_version_as_committed() { + 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 current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + let uncommitted = 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, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + backend + .seed_object(bucket, &uncommitted, br#"{"metadata-log":[]}"#.to_vec()) + .await; + + let report = store.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap(); + + assert_eq!(report.current_metadata_status, TableMetadataPointerStatus::Valid); + assert_eq!(report.catalog.table.metadata_location, current); + assert_eq!(report.orphan_metadata_candidate_locations, vec![uncommitted]); + } + #[tokio::test] async fn object_table_catalog_store_commits_with_token_match_and_writes_log() { let backend = TestCatalogObjectBackend::default();