diff --git a/rustfs/src/table_catalog/iceberg/validation.rs b/rustfs/src/table_catalog/iceberg/validation.rs index 74e3a2262..7cac23ae3 100644 --- a/rustfs/src/table_catalog/iceberg/validation.rs +++ b/rustfs/src/table_catalog/iceberg/validation.rs @@ -270,6 +270,29 @@ pub(crate) fn table_data_plane_resource_from_entry(table: TableEntry, warehouse_ } } +pub(crate) fn table_data_plane_resource_from_warehouse_index( + index: &TableWarehouseIndexEntry, +) -> TableCatalogStoreResult { + if index.table_bucket.is_empty() || index.table_id.is_empty() { + return Err(TableCatalogStoreError::Invalid( + "warehouse index has an empty table bucket or table id".to_string(), + )); + } + parse_namespace_for_store(&index.namespace)?; + parse_table_for_store(&index.table)?; + let normalized = normalize_warehouse_object_prefix(&index.warehouse_object_prefix, Some(WAREHOUSE_INDEX_MAX_PREFIX_DEPTH))?; + if normalized != index.warehouse_object_prefix { + return Err(TableCatalogStoreError::Invalid("warehouse index prefix is not canonical".to_string())); + } + Ok(TableDataPlaneResource { + table_bucket: index.table_bucket.clone(), + namespace: index.namespace.clone(), + table: index.table.clone(), + table_id: index.table_id.clone(), + warehouse_object_prefix: index.warehouse_object_prefix.clone(), + }) +} + pub(crate) async fn table_data_plane_resource_for_object( store: &S, bucket: &str, diff --git a/rustfs/src/table_catalog/model.rs b/rustfs/src/table_catalog/model.rs index 6fdc605e6..939a23027 100644 --- a/rustfs/src/table_catalog/model.rs +++ b/rustfs/src/table_catalog/model.rs @@ -258,10 +258,11 @@ pub(crate) struct TableWarehouseIndexStateEntry { pub(super) state: TableCatalogEntryState, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] pub(crate) enum WarehouseIndexReservation { Created, AlreadyReserved, + Replaced(TableWarehouseIndexEntry), } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index 26b90077e..f80db59d4 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -239,6 +239,12 @@ struct ActiveNamespaceEvidence { explicit_entry: Option, } +enum TableWarehouseIndexResolution { + Active(TableDataPlaneResource), + Deleted(TableDataPlaneResource), + Missing, +} + #[derive(Clone)] pub(crate) struct ObjectTableCatalogStore { pub(in crate::table_catalog) backend: B, @@ -1272,15 +1278,14 @@ where Ok(()) } - async fn replace_stale_table_warehouse_index( + async fn replace_table_warehouse_index( &self, object: &str, stale: &TableWarehouseIndexEntry, replacement: &TableWarehouseIndexEntry, - reason: &'static str, ) -> TableCatalogStoreResult { let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), object).await?; - let Some((current, _)) = self + let Some((current, current_etag)) = self .read_entry_unlocked::(self.catalog_bucket(), object) .await? else { @@ -1290,9 +1295,16 @@ where if current != *stale { return Ok(false); } - self.delete_warehouse_index_object_unlocked(object, stale, reason).await?; - self.write_entry_unlocked(self.catalog_bucket(), object, replacement, TableCatalogPutPrecondition::IfAbsent) - .await?; + let current_etag = current_etag + .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog warehouse index has no etag: {object}")))?; + // A missing index would expose residual data during failed prefix reuse. + self.write_entry_unlocked( + self.catalog_bucket(), + object, + replacement, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await?; Ok(true) } @@ -1398,11 +1410,8 @@ where index.warehouse_object_prefix ))); } - if self - .replace_stale_table_warehouse_index(&object, &existing, &index, "stale reservation conflict") - .await? - { - return Ok(WarehouseIndexReservation::Created); + if self.replace_table_warehouse_index(&object, &existing, &index).await? { + return Ok(WarehouseIndexReservation::Replaced(existing)); } } Err(err) => return Err(err), @@ -1422,12 +1431,12 @@ where .map_err(|err| TableCatalogStoreError::Internal(format!("failed to delete stale warehouse index {object}: {err}"))) } - async fn fail_closed_for_broken_warehouse_index( + async fn fail_closed_for_broken_warehouse_index( &self, object: &str, index: &TableWarehouseIndexEntry, reason: &'static str, - ) -> TableCatalogStoreResult> { + ) -> TableCatalogStoreResult> { Err(TableCatalogStoreError::Internal(format!( "active warehouse index {object} for {}/{}/{} ({}) is inconsistent: {reason}", index.table_bucket, index.namespace, index.table, index.table_id @@ -1438,11 +1447,15 @@ where &self, index_object: &str, index: TableWarehouseIndexEntry, - ) -> TableCatalogStoreResult> { + ) -> TableCatalogStoreResult> { validate_table_warehouse_index_entry_object(&self.paths, index_object, &index)?; + if index.state == TableCatalogEntryState::Deleted { + let resource = table_data_plane_resource_from_warehouse_index(&index)?; + return Ok(Some((TableCatalogEntryState::Deleted, resource))); + } if index.state != TableCatalogEntryState::Active { return Err(TableCatalogStoreError::Internal(format!( - "warehouse index {index_object} for {}/{}/{} is inactive while the index is authoritative", + "warehouse index {index_object} for {}/{}/{} has an invalid transient state while the index is authoritative", index.table_bucket, index.namespace, index.table ))); } @@ -1472,7 +1485,10 @@ where .fail_closed_for_broken_warehouse_index(index_object, &index, "referenced table identity changed") .await; } - Ok(Some(table_data_plane_resource_from_entry(table, current_prefix))) + Ok(Some(( + TableCatalogEntryState::Active, + table_data_plane_resource_from_entry(table, current_prefix), + ))) } async fn read_warehouse_index_state_unlocked(&self, table_bucket: &str) -> TableCatalogStoreResult { @@ -1488,17 +1504,41 @@ where table_warehouse_index_state_ready(&state, table_bucket) } - async fn delete_created_table_warehouse_index( + async fn rollback_table_warehouse_index_reservation( &self, entry: &TableEntry, reservation: WarehouseIndexReservation, reason: &'static str, ) { - if reservation != WarehouseIndexReservation::Created { - return; + let result = async { + match reservation { + WarehouseIndexReservation::AlreadyReserved => Ok(()), + WarehouseIndexReservation::Created => self.delete_table_warehouse_index(entry).await, + WarehouseIndexReservation::Replaced(previous) => { + let index = table_warehouse_index_entry(entry)?; + // An error response can follow a committed catalog write. + if let Some(current) = self + .load_table_entry(&entry.table_bucket, &entry.namespace, &entry.table) + .await? + && table_warehouse_index_entry(¤t)? == index + { + return Ok(()); + } + self.replace_table_warehouse_index( + &self + .paths + .warehouse_index_entry_path(&index.table_bucket, &index.warehouse_object_prefix), + &index, + &previous, + ) + .await + .map(|_| ()) + } + } } + .await; let warehouse_object_prefix = table_warehouse_object_prefix(entry).ok(); - if let Err(err) = self.delete_table_warehouse_index(entry).await { + if let Err(err) = result { tracing::warn!( table_bucket = %entry.table_bucket, namespace = %entry.namespace, @@ -1527,28 +1567,46 @@ where .map(|_| ()) } - async fn delete_owned_table_warehouse_index_for_drop(&self, entry: &TableEntry) -> TableCatalogStoreResult<()> { + pub(in crate::table_catalog) async fn tombstone_table_warehouse_index_for_drop( + &self, + entry: &TableEntry, + replace_deleted_owner: bool, + ) -> TableCatalogStoreResult<()> { let index = table_warehouse_index_entry(entry)?; + let mut tombstone = index.clone(); + tombstone.state = TableCatalogEntryState::Deleted; let object = self .paths .warehouse_index_entry_path(&index.table_bucket, &index.warehouse_object_prefix); validate_table_warehouse_index_entry_object(&self.paths, &object, &index)?; let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?; - let Some((current, _)) = self + let Some((current, current_etag)) = self .read_entry_unlocked::(self.catalog_bucket(), &object) .await? else { - return Ok(()); + return self + .write_entry_unlocked(self.catalog_bucket(), &object, &tombstone, TableCatalogPutPrecondition::IfAbsent) + .await; }; validate_table_warehouse_index_entry_object(&self.paths, &object, ¤t)?; - if current != index { + if current == tombstone { + return Ok(()); + } + if current != index && !(replace_deleted_owner && current.state == TableCatalogEntryState::Deleted) { return Err(TableCatalogStoreError::Conflict(format!( "table warehouse index owner changed before drop: {}", index.warehouse_object_prefix ))); } - self.delete_warehouse_index_object_unlocked(&object, &index, "table warehouse index owner dropped") - .await + let current_etag = current_etag + .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog warehouse index has no etag: {object}")))?; + self.write_entry_unlocked( + self.catalog_bucket(), + &object, + &tombstone, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await } async fn restore_table_warehouse_index_after_failed_drop(&self, entry: &TableEntry, reason: &'static str) { @@ -1592,8 +1650,10 @@ where &self, table_bucket: &str, object: &str, - ) -> TableCatalogStoreResult> { - let mut matched: Option = None; + ) -> TableCatalogStoreResult { + let mut active: Option = None; + let mut deleted: Option = None; + let mut deleted_overlap = None; for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) { let index_object = self.paths.warehouse_index_entry_path(table_bucket, warehouse_object_prefix); let Some((index, _)) = self @@ -1607,19 +1667,107 @@ where "warehouse index entry does not match indexed prefix: {index_object}" ))); } - if let Some(resource) = self + if let Some((state, resource)) = self .resolve_table_data_plane_resource_from_index_entry(&index_object, index) .await? { + let matched = match &state { + TableCatalogEntryState::Active => &mut active, + TableCatalogEntryState::Deleted => { + if let Some(current) = deleted.as_ref() { + deleted_overlap = + Some((current.warehouse_object_prefix.clone(), resource.warehouse_object_prefix.clone())); + continue; + } + &mut deleted + } + TableCatalogEntryState::Renaming | TableCatalogEntryState::Deleting => { + unreachable!("transient indexes are rejected above") + } + }; if let Some(current) = matched.as_ref() { return Err(TableCatalogStoreError::Invalid(format!( "object {object} matches overlapping active table warehouse indexes {} and {}", current.warehouse_object_prefix, resource.warehouse_object_prefix ))); } - matched = Some(resource); + *matched = Some(resource); } } + if let Some(resource) = active { + Ok(TableWarehouseIndexResolution::Active(resource)) + } else if let Some((left, right)) = deleted_overlap { + Err(TableCatalogStoreError::Invalid(format!( + "object {object} matches overlapping deleted table warehouse indexes {left} and {right}" + ))) + } else if let Some(resource) = deleted { + Ok(TableWarehouseIndexResolution::Deleted(resource)) + } else { + Ok(TableWarehouseIndexResolution::Missing) + } + } + + async fn resolve_table_data_plane_resource_with_scan( + &self, + table_bucket: &str, + object: &str, + ) -> TableCatalogStoreResult> { + match self + .resolve_table_data_plane_resource_from_index(table_bucket, object) + .await? + { + TableWarehouseIndexResolution::Active(resource) => Ok(Some(resource)), + TableWarehouseIndexResolution::Deleted(tombstone) => { + Ok(scan_table_data_plane_resource_for_object(self, table_bucket, object) + .await? + .or(Some(tombstone))) + } + TableWarehouseIndexResolution::Missing => scan_table_data_plane_resource_for_object(self, table_bucket, object).await, + } + } + + pub(in crate::table_catalog) async fn resolve_deleted_table_data_plane_resource_from_index( + &self, + table_bucket: &str, + object: &str, + ) -> TableCatalogStoreResult> { + let mut matched: Option = None; + for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) { + let index_object = self.paths.warehouse_index_entry_path(table_bucket, warehouse_object_prefix); + let Some((index, _)) = self + .read_entry::(self.catalog_bucket(), &index_object) + .await? + else { + continue; + }; + validate_table_warehouse_index_entry_object(&self.paths, &index_object, &index)?; + if index.table_bucket != table_bucket || index.warehouse_object_prefix != warehouse_object_prefix { + return Err(TableCatalogStoreError::Invalid(format!( + "warehouse index entry does not match indexed prefix: {index_object}" + ))); + } + match &index.state { + TableCatalogEntryState::Active => { + return Err(TableCatalogStoreError::Internal(format!( + "active warehouse index {index_object} is absent from the authoritative strong catalog" + ))); + } + TableCatalogEntryState::Deleted => {} + TableCatalogEntryState::Renaming | TableCatalogEntryState::Deleting => { + return Err(TableCatalogStoreError::Internal(format!( + "warehouse index {index_object} has an incomplete transient state" + ))); + } + } + let resource = table_data_plane_resource_from_warehouse_index(&index)?; + if let Some(current) = matched.as_ref() { + return Err(TableCatalogStoreError::Invalid(format!( + "object {object} matches overlapping deleted table warehouse indexes {} and {}", + current.warehouse_object_prefix, resource.warehouse_object_prefix + ))); + } + matched = Some(resource); + } Ok(matched) } @@ -1872,7 +2020,7 @@ where if !publication.holds_table_bucket(&entry.table_bucket) || !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table) { - self.delete_created_table_warehouse_index(&entry, reservation, "table publication fence lost") + self.rollback_table_warehouse_index_reservation(&entry, reservation, "table publication fence lost") .await; return Err(TableCatalogStoreError::Internal( "table registration publication fence was lost before catalog update".to_string(), @@ -1882,7 +2030,7 @@ where .write_entry_unlocked(self.catalog_bucket(), &table_path, &entry, precondition) .await; if result.is_err() { - self.delete_created_table_warehouse_index(&entry, reservation, "table entry write failed") + self.rollback_table_warehouse_index_reservation(&entry, reservation, "table entry write failed") .await; } result @@ -5119,29 +5267,17 @@ where let read_version = self.table_data_plane_read_version(table_bucket).await?; let resource = if self.warehouse_index_ready(table_bucket).await? { - match self - .resolve_table_data_plane_resource_from_index(table_bucket, object) - .await? - { - Some(resource) => Ok(Some(resource)), - None => scan_table_data_plane_resource_for_object(self, table_bucket, object).await, - } + self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await } else { match self.backfill_table_warehouse_index(table_bucket).await { - Ok(()) => match self - .resolve_table_data_plane_resource_from_index(table_bucket, object) - .await? - { - Some(resource) => Ok(Some(resource)), - None => scan_table_data_plane_resource_for_object(self, table_bucket, object).await, - }, + Ok(()) => self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await, Err(err @ TableCatalogStoreError::Internal(_)) => { tracing::warn!( table_bucket = %table_bucket, error = %err, "failed to backfill table warehouse index; falling back to catalog scan" ); - scan_table_data_plane_resource_for_object(self, table_bucket, object).await + self.resolve_table_data_plane_resource_with_scan(table_bucket, object).await } Err(err) => Err(err), } @@ -5524,7 +5660,7 @@ where } .await; if let Err(err) = staged_write_result { - self.delete_created_table_warehouse_index(&next, reservation, "commit staging failed") + self.rollback_table_warehouse_index_reservation(&next, reservation, "commit staging failed") .await; return table_commit_result( &request.table_bucket, @@ -5540,7 +5676,7 @@ where if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table) || (warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket)) { - self.delete_created_table_warehouse_index(&next, reservation, "table publication fence lost") + self.rollback_table_warehouse_index_reservation(&next, reservation, "table publication fence lost") .await; return table_commit_result( &request.table_bucket, @@ -5566,7 +5702,7 @@ where .await; record_table_commit_cas_result(&request.operation, cas_started, &cas_result); if let Err(err) = cas_result { - self.delete_created_table_warehouse_index(&next, reservation, "table pointer CAS failed") + self.rollback_table_warehouse_index_reservation(&next, reservation, "table pointer CAS failed") .await; return table_commit_result( &request.table_bucket, @@ -5632,7 +5768,7 @@ where table.as_str() ))); }; - self.delete_owned_table_warehouse_index_for_drop(&entry).await?; + self.tombstone_table_warehouse_index_for_drop(&entry, false).await?; if !publication.holds_table_bucket(table_bucket) || !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str()) { diff --git a/rustfs/src/table_catalog/store/strong.rs b/rustfs/src/table_catalog/store/strong.rs index 43f17aee5..455e40d64 100644 --- a/rustfs/src/table_catalog/store/strong.rs +++ b/rustfs/src/table_catalog/store/strong.rs @@ -2416,26 +2416,32 @@ where } self.hydrate_state().await?; - let state = self.state.lock().await; - self.require_data_plane_ready_locked(&state, table_bucket)?; + let active_resource = { + let state = self.state.lock().await; + self.require_data_plane_ready_locked(&state, table_bucket)?; - let Some(bucket_index) = state.warehouse_index.get(table_bucket) else { - return Ok(None); - }; - - for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) { - if let Some(table_key) = bucket_index.get(warehouse_object_prefix) { - Self::ensure_identifier_is_unambiguous_locked(&state, table_key)?; - let Some(table) = state.tables.get(table_key) else { - continue; - }; - return Ok(Some(table_data_plane_resource_from_entry( - table.clone(), - warehouse_object_prefix.to_string(), - ))); + let mut resource = None; + if let Some(bucket_index) = state.warehouse_index.get(table_bucket) { + for warehouse_object_prefix in warehouse_index_candidate_prefixes(object) { + if let Some(table_key) = bucket_index.get(warehouse_object_prefix) { + Self::ensure_identifier_is_unambiguous_locked(&state, table_key)?; + let Some(table) = state.tables.get(table_key) else { + continue; + }; + resource = Some(table_data_plane_resource_from_entry(table.clone(), warehouse_object_prefix.to_string())); + break; + } + } } + resource + }; + if active_resource.is_some() { + return Ok(active_resource); } - Ok(None) + + ObjectTableCatalogStore::new(self.object_backend.clone()) + .resolve_deleted_table_data_plane_resource_from_index(table_bucket, object) + .await } async fn resolve_table_metadata_data_plane_resource( @@ -2674,11 +2680,22 @@ where let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; let key = Self::table_key(table_bucket, &namespace, &table); + let dropped = { + let state = self.state.lock().await; + state.tables.get(&key).cloned().ok_or_else(|| { + TableCatalogStoreError::NotFound(format!("table {}/{}/{}", table_bucket, namespace.public_name(), table.as_str())) + })? + }; + if dropped.state == TableCatalogEntryState::Active { + ObjectTableCatalogStore::new(self.object_backend.clone()) + .tombstone_table_warehouse_index_for_drop(&dropped, true) + .await?; + } let (snapshot, precondition, postcondition) = { let state = self.state.lock().await; - if !state.tables.contains_key(&key) { - return Err(TableCatalogStoreError::NotFound(format!( - "table {}/{}/{}", + if state.tables.get(&key) != Some(&dropped) { + return Err(TableCatalogStoreError::Conflict(format!( + "table changed while preparing drop: {}/{}/{}", table_bucket, namespace.public_name(), table.as_str() diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index b1e43446d..8289f1880 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -4534,7 +4534,7 @@ async fn table_data_plane_resource_scans_when_an_index_disappears_during_backfil } #[tokio::test] -async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() { +async fn active_catalog_owner_takes_precedence_over_a_warehouse_tombstone() { let backend = TestCatalogObjectBackend::default(); let store = ObjectTableCatalogStore::new(backend.clone()); let bucket = "analytics"; @@ -4544,7 +4544,7 @@ async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() { let prefix = "tables/table-id/"; seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; - let inactive = TableWarehouseIndexEntry { + let tombstone = TableWarehouseIndexEntry { version: TABLE_CATALOG_ENTRY_VERSION, table_bucket: bucket.to_string(), namespace: namespace.public_name(), @@ -4557,16 +4557,58 @@ async fn table_data_plane_resource_fails_closed_for_an_inactive_ready_index() { .write_entry( store.catalog_bucket(), &store.paths.warehouse_index_entry_path(bucket, prefix), - &inactive, + &tombstone, TableCatalogPutPrecondition::Any, ) .await - .expect("inactive warehouse index should be seeded"); + .expect("warehouse tombstone should be seeded"); - assert_matches!( - table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet").await, - Err(TableCatalogStoreError::Internal(message)) if message.contains("inactive while the index is authoritative") - ); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") + .await + .expect("active catalog owner should resolve ahead of the tombstone") + .expect("active catalog owner should retain table-aware protection"); + + assert_eq!(resource.table_id, "table-id"); + assert_eq!(resource.warehouse_object_prefix, prefix); +} + +#[tokio::test] +async fn table_data_plane_resource_fails_closed_for_a_transient_ready_index() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let prefix = "tables/table-id/"; + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + for state in [TableCatalogEntryState::Renaming, TableCatalogEntryState::Deleting] { + let index = TableWarehouseIndexEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + table_id: "table-id".to_string(), + warehouse_object_prefix: prefix.to_string(), + state, + }; + store + .write_entry( + store.catalog_bucket(), + &store.paths.warehouse_index_entry_path(bucket, prefix), + &index, + TableCatalogPutPrecondition::Any, + ) + .await + .expect("transient warehouse index should be seeded"); + + assert_matches!( + table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet").await, + Err(TableCatalogStoreError::Internal(message)) + if message.contains("invalid transient state while the index is authoritative") + ); + } } #[tokio::test] @@ -5391,19 +5433,130 @@ async fn object_table_catalog_store_rolls_back_warehouse_index_when_table_entry_ } #[tokio::test] -async fn object_table_catalog_store_keeps_table_when_drop_index_delete_fails() { +async fn object_catalog_prefix_reuse_preserves_protection_on_write_failure() { + for failure in ["index", "table", "rollback", "ambiguous"] { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let first = IdentifierSegment::parse("orders").unwrap(); + let second = IdentifierSegment::parse("returns").unwrap(); + let current = default_table_metadata_file_path(&namespace, &first, "00001.metadata.json"); + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &first, current).await; + store + .drop_table(bucket, &namespace.public_name(), first.as_str()) + .await + .unwrap(); + + let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + let table_path = store.paths.table_entry_path(bucket, &namespace, &second); + let index_attempts = backend.put_attempt_count(RUSTFS_META_BUCKET, &index_path).await; + if failure == "index" { + backend + .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, index_attempts + 2) + .await; + } else if failure == "ambiguous" { + backend.fail_after_next_put(RUSTFS_META_BUCKET, &table_path).await; + } else { + backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await; + if failure == "rollback" { + backend + .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, index_attempts + 3) + .await; + } + } + + let metadata = default_table_metadata_file_path(&namespace, &second, "00001.metadata.json"); + let mut entry = test_table_entry(bucket, &namespace, &second, metadata); + entry.table_id = "replacement-table-id".to_string(); + entry.warehouse_location = format!("s3://{bucket}/tables/table-id"); + assert_matches!(store.create_table(entry.clone()).await, Err(TableCatalogStoreError::Internal(_))); + assert_eq!( + store + .load_table(bucket, &namespace.public_name(), second.as_str()) + .await + .unwrap() + .is_some(), + failure == "ambiguous" + ); + + let restarted = ObjectTableCatalogStore::new(backend.clone()); + let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet").await; + if failure == "rollback" { + assert_matches!(resource, Err(TableCatalogStoreError::Internal(_))); + } else { + let resource = resource + .expect("lookup should succeed") + .expect("failed prefix reuse must retain protection"); + assert_eq!( + resource.table_id, + if failure == "ambiguous" { + "replacement-table-id" + } else { + "table-id" + }, + "failure at {failure}" + ); + } + let (index, _) = restarted + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .unwrap() + .expect("failed prefix reuse must not remove the warehouse index"); + assert_eq!( + index.state, + if matches!(failure, "rollback" | "ambiguous") { + TableCatalogEntryState::Active + } else { + TableCatalogEntryState::Deleted + } + ); + + if failure == "rollback" { + let third = IdentifierSegment::parse("retry_returns").unwrap(); + let mut retry = entry.clone(); + retry.table = third.as_str().to_string(); + retry.table_id = "retry-table-id".to_string(); + retry.metadata_location = default_table_metadata_file_path(&namespace, &third, "00001.metadata.json"); + backend + .fail_next_put(RUSTFS_META_BUCKET, &store.paths.table_entry_path(bucket, &namespace, &third)) + .await; + assert_matches!(restarted.create_table(retry).await, Err(TableCatalogStoreError::Internal(_))); + assert_matches!( + table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet").await, + Err(TableCatalogStoreError::Internal(_)) + ); + } + if failure != "ambiguous" { + restarted + .create_table(entry) + .await + .expect("prefix reuse should remain retryable"); + } + restarted + .drop_table(bucket, &namespace.public_name(), second.as_str()) + .await + .expect("replacement table should remain droppable"); + let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet") + .await + .unwrap() + .expect("replacement tombstone should protect the reused prefix"); + assert_eq!(resource.table_id, "replacement-table-id"); + } +} + +#[tokio::test] +async fn object_table_catalog_store_keeps_table_when_drop_index_tombstone_write_fails() { let backend = TestCatalogObjectBackend::default(); let store = ObjectTableCatalogStore::new(backend.clone()); let bucket = "analytics"; let namespace = Namespace::parse("sales").expect("namespace should parse"); let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); - let table_path = store.paths.table_entry_path(bucket, &namespace, &table); let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; - backend.fail_delete_attempt(RUSTFS_META_BUCKET, &index_path, 1).await; - backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await; + backend.fail_next_put(RUSTFS_META_BUCKET, &index_path).await; let error = store .drop_table(bucket, &namespace.public_name(), table.as_str()) @@ -5467,6 +5620,50 @@ async fn object_table_catalog_store_rejects_drop_when_warehouse_index_owner_chan assert_eq!(retained_index, conflicting_index); } +#[tokio::test] +async fn object_table_catalog_store_rejects_drop_when_deleted_warehouse_index_owner_changed() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let index_path = store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + let conflicting_index = TableWarehouseIndexEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.to_string(), + namespace: "finance".to_string(), + table: "returns".to_string(), + table_id: "other-table-id".to_string(), + warehouse_object_prefix: "tables/table-id/".to_string(), + state: TableCatalogEntryState::Deleted, + }; + store + .write_entry(store.catalog_bucket(), &index_path, &conflicting_index, TableCatalogPutPrecondition::Any) + .await + .expect("conflicting warehouse tombstone should be seeded"); + + assert_matches!( + store.drop_table(bucket, &namespace.public_name(), table.as_str()).await, + Err(TableCatalogStoreError::Conflict(message)) if message.contains("owner changed") + ); + assert!( + store + .load_table(bucket, &namespace.public_name(), table.as_str()) + .await + .expect("retained table lookup should succeed") + .is_some() + ); + let (retained_index, _) = store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("conflicting warehouse tombstone lookup should succeed") + .expect("conflicting warehouse tombstone should remain present"); + assert_eq!(retained_index, conflicting_index); +} + #[tokio::test] async fn object_table_catalog_store_drops_table_when_warehouse_index_is_missing() { let backend = TestCatalogObjectBackend::default(); @@ -5506,6 +5703,76 @@ async fn object_table_catalog_store_drops_table_when_warehouse_index_is_missing( .expect("dropped table lookup should succeed") .is_none() ); + let (tombstone, _) = store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("warehouse tombstone lookup should succeed") + .expect("warehouse tombstone should be created"); + assert_eq!(tombstone.state, TableCatalogEntryState::Deleted); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") + .await + .expect("dropped table data-plane lookup should succeed") + .expect("dropped warehouse prefix should remain protected"); + assert_eq!(resource.table_id, "table-id"); +} + +#[tokio::test] +async fn active_table_warehouse_prefix_takes_precedence_over_a_deleted_parent() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let parent = IdentifierSegment::parse("orders").expect("table should parse"); + let child = IdentifierSegment::parse("returns").expect("table should parse"); + let parent_metadata = default_table_metadata_file_path(&namespace, &parent, "00001.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &parent, parent_metadata).await; + store + .drop_table(bucket, &namespace.public_name(), parent.as_str()) + .await + .expect("parent table should be dropped"); + + let child_metadata = default_table_metadata_file_path(&namespace, &child, "00001.metadata.json"); + let mut child_entry = test_table_entry(bucket, &namespace, &child, child_metadata); + child_entry.table_id = "child-table-id".to_string(); + child_entry.warehouse_location = format!("s3://{bucket}/tables/table-id/child"); + store.create_table(child_entry).await.expect("child table should be created"); + + let child_resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/data/file.parquet") + .await + .expect("child data-plane lookup should succeed") + .expect("active child table should protect its prefix"); + assert_eq!(child_resource.table_id, "child-table-id"); + + let parent_resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/orphan/file.parquet") + .await + .expect("deleted parent data-plane lookup should succeed") + .expect("deleted parent should keep protecting its remaining prefix"); + assert_eq!(parent_resource.table_id, "table-id"); + + store + .drop_table(bucket, &namespace.public_name(), child.as_str()) + .await + .unwrap(); + assert_matches!( + table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/data/file.parquet").await, + Err(TableCatalogStoreError::Invalid(message)) if message.contains("overlapping deleted table warehouse indexes") + ); + let grandchild = IdentifierSegment::parse("nested_returns").unwrap(); + let mut entry = test_table_entry( + bucket, + &namespace, + &grandchild, + default_table_metadata_file_path(&namespace, &grandchild, "00001.metadata.json"), + ); + entry.table_id = "grandchild-table-id".to_string(); + entry.warehouse_location = format!("s3://{bucket}/tables/table-id/child/nested"); + store.create_table(entry).await.unwrap(); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/child/nested/data/file.parquet") + .await + .unwrap() + .expect("active owner should take precedence over both tombstones"); + assert_eq!(resource.table_id, "grandchild-table-id"); } #[tokio::test] @@ -5538,6 +5805,7 @@ async fn object_table_catalog_store_restores_index_when_table_entry_delete_fails .expect("restored warehouse index lookup should succeed") .expect("warehouse index should be restored"); assert_eq!(restored_index.table_id, "table-id"); + assert_eq!(restored_index.state, TableCatalogEntryState::Active); let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") .await .expect("restored index lookup should succeed") @@ -5546,7 +5814,7 @@ async fn object_table_catalog_store_restores_index_when_table_entry_delete_fails } #[tokio::test] -async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_fails() { +async fn object_table_catalog_store_uses_tombstone_when_drop_index_restore_fails() { let backend = TestCatalogObjectBackend::default(); let store = ObjectTableCatalogStore::new(backend.clone()); let bucket = "analytics"; @@ -5558,7 +5826,10 @@ async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_f seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; backend.fail_delete_attempt(RUSTFS_META_BUCKET, &table_path, 1).await; - backend.fail_next_put(RUSTFS_META_BUCKET, &index_path).await; + let restore_attempt = backend.put_attempt_count(RUSTFS_META_BUCKET, &index_path).await + 2; + backend + .fail_put_attempt(RUSTFS_META_BUCKET, &index_path, restore_attempt) + .await; assert_matches!( store.drop_table(bucket, &namespace.public_name(), table.as_str()).await, @@ -5570,18 +5841,16 @@ async fn object_table_catalog_store_falls_back_to_scan_when_drop_index_restore_f .expect("retained table lookup should succeed") .expect("table entry should remain present"); assert_eq!(retained.table_id, "table-id"); - assert!( - store - .read_entry::(RUSTFS_META_BUCKET, &index_path) - .await - .expect("warehouse index lookup should succeed") - .is_none(), - "the injected index restore failure must leave the scan fallback under test" - ); + let (tombstone, _) = store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("warehouse index lookup should succeed") + .expect("the failed restore should retain the warehouse tombstone"); + assert_eq!(tombstone.state, TableCatalogEntryState::Deleted); let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") .await - .expect("catalog scan fallback should succeed") - .expect("catalog scan fallback should keep data-plane protection"); + .expect("tombstone fallback should succeed") + .expect("tombstone fallback should keep data-plane protection"); assert_eq!(resource.table, "orders"); } @@ -5609,12 +5878,11 @@ async fn object_table_catalog_store_accepts_ambiguous_delete_when_table_is_absen .expect("dropped table lookup should succeed") .is_none() ); - assert!( - table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") - .await - .expect("dropped table data-plane lookup should succeed") - .is_none() - ); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/part-00001.parquet") + .await + .expect("dropped table data-plane lookup should succeed") + .expect("dropped warehouse prefix should remain protected"); + assert_eq!(resource.table_id, "table-id"); } #[tokio::test] @@ -12609,6 +12877,196 @@ async fn strong_catalog_table_metadata_data_plane_resource_resolves_only_current ); } +#[tokio::test] +async fn strong_catalog_drop_keeps_warehouse_tombstone_after_restart() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + store + .create_table(test_table_entry(bucket, &namespace, &table, current)) + .await + .expect("table should be created"); + store + .drop_table(bucket, &namespace.public_name(), table.as_str()) + .await + .expect("table should be dropped"); + + let restarted = StrongTableCatalogStore::new(backend.clone()); + assert!( + restarted + .load_table(bucket, &namespace.public_name(), table.as_str()) + .await + .expect("dropped table lookup should succeed") + .is_none() + ); + let resource = table_data_plane_resource_for_object(&restarted, bucket, "tables/table-id/data/file.parquet") + .await + .expect("dropped table data-plane lookup should succeed") + .expect("dropped warehouse prefix should remain protected"); + assert_eq!(resource.table_id, "table-id"); + + let object_store = ObjectTableCatalogStore::new(backend); + let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + let (tombstone, _) = object_store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("warehouse tombstone lookup should succeed") + .expect("warehouse tombstone should remain durable"); + assert_eq!(tombstone.state, TableCatalogEntryState::Deleted); +} + +#[tokio::test] +async fn strong_catalog_failed_drop_keeps_active_table_ahead_of_tombstone() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + store + .create_table(test_table_entry(bucket, &namespace, &table, current)) + .await + .expect("table should be created"); + let snapshot_path = StrongTableCatalogStore::::snapshot_object_path(); + backend.fail_next_put(RUSTFS_META_BUCKET, &snapshot_path).await; + + assert_matches!( + store.drop_table(bucket, &namespace.public_name(), table.as_str()).await, + Err(TableCatalogStoreError::Internal(_)) + ); + assert!( + store + .load_table(bucket, &namespace.public_name(), table.as_str()) + .await + .expect("retained table lookup should succeed") + .is_some() + ); + let resource = table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/file.parquet") + .await + .expect("retained table data-plane lookup should succeed") + .expect("active table should remain protected after the failed drop"); + assert_eq!(resource.table_id, "table-id"); + + let object_store = ObjectTableCatalogStore::new(backend); + let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + let (tombstone, _) = object_store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("warehouse tombstone lookup should succeed") + .expect("failed snapshot write should retain the warehouse tombstone"); + assert_eq!(tombstone.state, TableCatalogEntryState::Deleted); +} + +#[tokio::test] +async fn strong_catalog_reused_warehouse_prefix_replaces_the_old_tombstone_on_drop() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let first = IdentifierSegment::parse("orders").expect("table should parse"); + let second = IdentifierSegment::parse("returns").expect("table should parse"); + let first_metadata = default_table_metadata_file_path(&namespace, &first, "00001.metadata.json"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + store + .create_table(test_table_entry(bucket, &namespace, &first, first_metadata)) + .await + .expect("first table should be created"); + store + .drop_table(bucket, &namespace.public_name(), first.as_str()) + .await + .expect("first table should be dropped"); + + let second_metadata = default_table_metadata_file_path(&namespace, &second, "00001.metadata.json"); + let mut second_entry = test_table_entry(bucket, &namespace, &second, second_metadata); + second_entry.table_id = "second-table-id".to_string(); + second_entry.warehouse_location = format!("s3://{bucket}/tables/table-id"); + store + .create_table(second_entry) + .await + .expect("second table should reuse the prefix"); + store + .drop_table(bucket, &namespace.public_name(), second.as_str()) + .await + .expect("second table should replace the old tombstone when dropped"); + + let object_store = ObjectTableCatalogStore::new(backend); + let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + let (tombstone, _) = object_store + .read_entry::(RUSTFS_META_BUCKET, &index_path) + .await + .expect("warehouse tombstone lookup should succeed") + .expect("warehouse tombstone should remain durable"); + assert_eq!(tombstone.state, TableCatalogEntryState::Deleted); + assert_eq!(tombstone.table_id, "second-table-id"); + assert_eq!(tombstone.table, "returns"); +} + +#[tokio::test] +async fn strong_catalog_fails_closed_for_an_unowned_active_external_warehouse_index() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + + let object_store = ObjectTableCatalogStore::new(backend); + let index_path = object_store.paths.warehouse_index_entry_path(bucket, "tables/table-id/"); + let unowned = TableWarehouseIndexEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.to_string(), + namespace: "sales".to_string(), + table: "orders".to_string(), + table_id: "table-id".to_string(), + warehouse_object_prefix: "tables/table-id/".to_string(), + state: TableCatalogEntryState::Active, + }; + object_store + .write_entry( + object_store.catalog_bucket(), + &index_path, + &unowned, + TableCatalogPutPrecondition::IfAbsent, + ) + .await + .expect("unowned active warehouse index should be seeded"); + + assert_matches!( + table_data_plane_resource_for_object(&store, bucket, "tables/table-id/data/file.parquet").await, + Err(TableCatalogStoreError::Internal(message)) if message.contains("authoritative strong catalog") + ); +} + #[tokio::test] async fn strong_catalog_data_plane_fails_closed_for_missing_bucket_snapshot() { let store = StrongTableCatalogStore::new(TestCatalogObjectBackend::default());