diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 1dfd412a3..309bd1573 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -64,7 +64,7 @@ catalog extension. | Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and only the supported OpenAPI REST paths in `endpoints`. RustFS administration, maintenance, migration, diagnostics, refs, and metadata-location extensions remain available but are not presented as standard Iceberg REST endpoints. | | Table bucket discovery | Supported | `PUT` and `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. | | Namespaces | Supported | Create, list, load, existence check, and drop namespace routes are registered on both catalog prefixes. List responses support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Namespace identifiers are limited to 512 ASCII characters so persisted paths and stateless continuation tokens remain bounded. | -| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Commit identifiers must match the URL resource; unknown requirements, updates, and snapshot operations fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. Standard statistics, partition statistics, and schema/spec cleanup updates are accepted. | +| Tables | Supported | Create, register, list, load, existence check, rename, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Object-backed rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication of the destination, source tombstone, and warehouse index; the source identifier is reusable only through an ETag-conditional tombstone replacement. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Commit identifiers must match the URL resource; unknown requirements, updates, and snapshot operations fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. Standard statistics, partition statistics, and schema/spec cleanup updates are accepted. | | Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. Externally supplied metadata transitions preserve monotonic column, partition, and sequence assignment watermarks and immutable definitions for retained schemas, partition specs, sort orders, and snapshots. Standard commits preserve the normal commit-token file name and use an immutable-table-scoped fallback when rename followed by source-name reuse would otherwise collide at the same generation and commit ID. The catalog does not advertise `idempotency-key-lifetime`; clients must treat standard mutation-wide `Idempotency-Key` semantics as unsupported. | | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose staged/finalization gaps and repair safe idempotency gaps without moving the table pointer. | | Snapshot refs | Supported | Refs can be listed, created or replaced, and deleted through catalog commits. `main` is protected and refs with explicit retention require forced delete. | diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index d958c8c9f..3b98f27ec 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -168,6 +168,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/credentials", "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}", "DELETE /v1/{prefix}/namespaces/{namespace}/tables/{table}", + "POST /v1/{prefix}/tables/rename", "GET /v1/{prefix}/namespaces/{namespace}/views", "POST /v1/{prefix}/namespaces/{namespace}/views", "GET /v1/{prefix}/namespaces/{namespace}/views/{view}", @@ -175,10 +176,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "POST /v1/{prefix}/namespaces/{namespace}/views/{view}", "DELETE /v1/{prefix}/namespaces/{namespace}/views/{view}", ]; -const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &[ - "POST /v1/{prefix}/namespaces/{namespace}/properties", - "POST /v1/{prefix}/tables/rename", -]; +const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &["POST /v1/{prefix}/namespaces/{namespace}/properties"]; static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {}; static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {}; @@ -2298,6 +2296,7 @@ fn table_bucket_entry_from_metadata_marker(bucket: &str) -> crate::table_catalog warehouse_root: format!("s3://{bucket}/"), state: crate::table_catalog::TableCatalogEntryState::Active, properties: BTreeMap::new(), + active_rename_id: None, created_at: None, updated_at: None, } diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 7e1520b87..0d68a36dc 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -438,7 +438,7 @@ fn catalog_config_response_lists_standard_rest_endpoints() { .endpoints .contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties") ); - assert!(!response.endpoints.contains(&"POST /v1/{prefix}/tables/rename")); + assert!(response.endpoints.contains(&"POST /v1/{prefix}/tables/rename")); assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog"); @@ -11120,6 +11120,7 @@ async fn seed_object_table_for_metadata_maintenance( warehouse_root: format!("s3://{bucket}/"), state: crate::table_catalog::TableCatalogEntryState::Active, properties: BTreeMap::new(), + active_rename_id: None, created_at: None, updated_at: None, }) diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index 687d18f11..c471457c0 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -86,6 +86,14 @@ impl ApiError { } } + pub fn service_unavailable() -> Self { + ApiError { + code: S3ErrorCode::ServiceUnavailable, + message: Self::error_code_to_message(&S3ErrorCode::ServiceUnavailable), + source: None, + } + } + pub fn invalid_request(message: impl std::fmt::Display) -> Self { ApiError { code: S3ErrorCode::InvalidRequest, diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 42775eabc..65125c03d 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -1598,7 +1598,11 @@ async fn table_data_plane_resource_for_request( error = %err, "failed to resolve table data-plane resource" ); - s3_error!(AccessDenied, "Access Denied") + if matches!(err, crate::table_catalog::TableCatalogStoreError::Unavailable(_)) { + S3Error::from(ApiError::service_unavailable()) + } else { + s3_error!(AccessDenied, "Access Denied") + } })?; let bucket_fence_key = (bucket.to_string(), crate::table_catalog::default_table_bucket_publication_lock_path()); let mut state = retained.state.lock(); diff --git a/rustfs/src/table_catalog/mod.rs b/rustfs/src/table_catalog/mod.rs index e82c9e825..dfdf982e9 100644 --- a/rustfs/src/table_catalog/mod.rs +++ b/rustfs/src/table_catalog/mod.rs @@ -101,6 +101,7 @@ pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1; )] pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1; pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1; +pub(crate) const TABLE_RENAME_INTENT_VERSION: u16 = 1; pub(crate) const TABLE_WAREHOUSE_INDEX_STATE_VERSION: u16 = 2; pub(crate) const TABLE_MAINTENANCE_CONFIG_VERSION: u16 = 1; pub(crate) const TABLE_EXTERNAL_CATALOG_BRIDGE_VERSION: u16 = 1; @@ -166,6 +167,7 @@ const COMMIT_LOG_ROOT: &str = "commits"; const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency"; const WAREHOUSE_INDEX_ROOT: &str = "warehouse-index"; const WAREHOUSE_INDEX_STATE_FILE: &str = "state.json"; +const TABLE_RENAME_ROOT: &str = "renames"; const WAREHOUSE_INDEX_MAX_PREFIX_DEPTH: usize = 64; const EXTERNAL_CATALOG_ROOT: &str = "external-catalog"; const EXTERNAL_CATALOG_BRIDGE_FILE: &str = "bridge.json"; diff --git a/rustfs/src/table_catalog/model.rs b/rustfs/src/table_catalog/model.rs index d091dd849..6fdc605e6 100644 --- a/rustfs/src/table_catalog/model.rs +++ b/rustfs/src/table_catalog/model.rs @@ -39,6 +39,8 @@ pub(crate) fn table_bucket_marker_json() -> Result, serde_json::Error> { #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableCatalogEntryState { Active, + /// Persisted only behind a rename intent so older readers reject the unknown state and fail closed. + Renaming, Deleting, Deleted, } @@ -53,10 +55,40 @@ pub(crate) struct TableBucketEntry { pub state: TableCatalogEntryState, #[serde(default)] pub properties: BTreeMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub active_rename_id: Option, pub created_at: Option, pub updated_at: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableRenameIntentState { + Prepared, + SourceFenced, + DestinationWritten, + SourceTombstoned, + IndexPublished, + DestinationPublished, + Completed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableRenameIntent { + pub version: u16, + pub rename_id: String, + pub table_bucket: String, + pub source: TableEntry, + pub destination: TableEntry, + pub source_etag: String, + pub destination_etag: Option, + pub warehouse_index_etag: String, + pub state: TableRenameIntentState, + pub created_at: String, + pub updated_at: String, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct NamespaceEntry { @@ -1225,6 +1257,7 @@ pub(crate) enum TableCatalogBackingMigrationStep { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "SCREAMING_SNAKE_CASE")] pub(crate) enum TableCatalogBackingMigrationBlocker { + TableRenameRecoveryRequired, CommitRecoveryRequired, CommitManualReviewRequired, WarehouseIndexBackfillRequired, diff --git a/rustfs/src/table_catalog/store/migration.rs b/rustfs/src/table_catalog/store/migration.rs index 398285f88..3bd77e2aa 100644 --- a/rustfs/src/table_catalog/store/migration.rs +++ b/rustfs/src/table_catalog/store/migration.rs @@ -301,6 +301,11 @@ where return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); }; validate_table_bucket_entry_object(&self.paths, &bucket_path, &table_bucket_entry)?; + if table_bucket_entry.active_rename_id.is_some() { + return Err(TableCatalogStoreError::Conflict(format!( + "table bucket {table_bucket} has a table rename requiring recovery" + ))); + } let mut namespaces = Vec::new(); let mut tables = Vec::new(); @@ -343,6 +348,9 @@ where ))); }; validate_table_entry_object(&self.paths, table_object, &table_entry)?; + if table_entry.state != TableCatalogEntryState::Active { + continue; + } for commit_object in self .backend @@ -595,9 +603,10 @@ where &self, table_bucket: &str, ) -> TableCatalogStoreResult { - if self.get_table_bucket(table_bucket).await?.is_none() { + let Some(table_bucket_entry) = self.get_table_bucket(table_bucket).await? else { return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); - } + }; + let rename_recovery_required = table_bucket_entry.active_rename_id.is_some(); if let Some((global_fence, _)) = self .read_entry::( self.catalog_bucket(), @@ -653,18 +662,19 @@ where continue; }; validate_table_entry_object(&self.paths, &object, &table)?; + if table.state != TableCatalogEntryState::Active { + continue; + } table_count = table_count.saturating_add(1); if !table_ids.insert(table.table_id.clone()) { duplicate_table_identity = true; } - if table.state == TableCatalogEntryState::Active { - active_table_identifiers.insert((table.namespace.clone(), table.table.clone())); - let warehouse_prefix = table_warehouse_object_prefix(&table)?; - warehouse_prefix_owners - .entry(warehouse_prefix) - .and_modify(|count| *count = count.saturating_add(1)) - .or_insert(1); - } + active_table_identifiers.insert((table.namespace.clone(), table.table.clone())); + let warehouse_prefix = table_warehouse_object_prefix(&table)?; + warehouse_prefix_owners + .entry(warehouse_prefix) + .and_modify(|count| *count = count.saturating_add(1)) + .or_insert(1); let recovery = self.table_commit_recovery_report_for_entry(&table, 0).await?; commit_log_count = commit_log_count.saturating_add(recovery.commits.len()); @@ -711,33 +721,41 @@ where let table_view_identifier_collision_count = active_table_identifiers.intersection(&active_view_identifiers).count(); let mut blockers = Vec::new(); let mut recommended_actions = Vec::new(); + if rename_recovery_required { + blockers.push(TableCatalogBackingMigrationBlocker::TableRenameRecoveryRequired); + recommended_actions.push(TableCatalogBackingMigrationAction::RunCatalogRecovery); + } if recovery_required_count > 0 { blockers.push(TableCatalogBackingMigrationBlocker::CommitRecoveryRequired); } if manual_review_count > 0 { blockers.push(TableCatalogBackingMigrationBlocker::CommitManualReviewRequired); } - if recovery_required_count > 0 || manual_review_count > 0 { + if (recovery_required_count > 0 || manual_review_count > 0) + && !recommended_actions.contains(&TableCatalogBackingMigrationAction::RunCatalogRecovery) + { recommended_actions.push(TableCatalogBackingMigrationAction::RunCatalogRecovery); } if !warehouse_index_ready { blockers.push(TableCatalogBackingMigrationBlocker::WarehouseIndexBackfillRequired); recommended_actions.push(TableCatalogBackingMigrationAction::BackfillWarehouseIndex); } - if conflicting_warehouse_prefix { + if conflicting_warehouse_prefix && !rename_recovery_required { blockers.push(TableCatalogBackingMigrationBlocker::DuplicateWarehousePrefix); recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDuplicateWarehousePrefixes); } - if duplicate_table_identity { + if duplicate_table_identity && !rename_recovery_required { blockers.push(TableCatalogBackingMigrationBlocker::DuplicateTableIdentity); recommended_actions.push(TableCatalogBackingMigrationAction::ReviewDuplicateTableIdentities); } - if table_view_identifier_collision_count > 0 { + if table_view_identifier_collision_count > 0 && !rename_recovery_required { blockers.push(TableCatalogBackingMigrationBlocker::TableViewIdentifierCollision); recommended_actions.push(TableCatalogBackingMigrationAction::ReviewTableViewIdentifierCollisions); } - let mut status = if manual_review_count > 0 + let mut status = if rename_recovery_required { + TableCatalogBackingMigrationStatus::RecoveryRequired + } else if manual_review_count > 0 || conflicting_warehouse_prefix || duplicate_table_identity || table_view_identifier_collision_count > 0 diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index 2dba0ad9a..9043db5ba 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -57,6 +57,9 @@ fn validate_table_bucket_entry(entry: &TableBucketEntry) -> TableCatalogStoreRes if entry.catalog_type != TABLE_BUCKET_CATALOG_TYPE { return Err(TableCatalogStoreError::Invalid("unsupported table bucket catalog type".to_string())); } + if entry.active_rename_id.as_ref().is_some_and(String::is_empty) { + return Err(TableCatalogStoreError::Invalid("active table rename id cannot be empty".to_string())); + } Ok(()) } @@ -984,6 +987,15 @@ impl TableCatalogObjectPaths { ) } + pub fn table_rename_intent_path(&self, table_bucket: &str, rename_id: &str) -> String { + format!( + "{}{}/{}.json", + self.table_bucket_root_prefix(table_bucket), + TABLE_RENAME_ROOT, + table_catalog_path_hash(rename_id) + ) + } + pub fn backing_migration_fence_path(&self, table_bucket: &str) -> String { format!( "{}{}/{}", @@ -1241,9 +1253,11 @@ where destination_table: &str, ) -> TableCatalogStoreResult<()> { match self { - Self::ObjectBacked(_) => Err(TableCatalogStoreError::Unsupported( - "table rename requires durable-strong catalog backing".to_string(), - )), + Self::ObjectBacked(store) => { + store + .rename_table(table_bucket, source_namespace, source_table, destination_namespace, destination_table) + .await + } Self::DurableStrong(store) => { store .rename_table(table_bucket, source_namespace, source_table, destination_namespace, destination_table) diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index bade7b02d..d696414a4 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -58,6 +58,65 @@ pub(super) fn validate_table_entry_object( Ok(namespace) } +fn validate_table_rename_intent_object( + paths: &TableCatalogObjectPaths, + object: &str, + intent: &TableRenameIntent, +) -> TableCatalogStoreResult<()> { + if intent.version != TABLE_RENAME_INTENT_VERSION + || intent.rename_id.is_empty() + || intent.source_etag.is_empty() + || intent.destination_etag.as_deref().is_some_and(str::is_empty) + || intent.warehouse_index_etag.is_empty() + || intent.created_at.is_empty() + || intent.updated_at.is_empty() + { + return Err(TableCatalogStoreError::Invalid( + "catalog table rename intent has invalid required fields".to_string(), + )); + } + if paths.table_rename_intent_path(&intent.table_bucket, &intent.rename_id) != object { + return Err(TableCatalogStoreError::Invalid( + "catalog table rename intent identity does not match its object path".to_string(), + )); + } + validate_table_entry_version_and_id(&intent.source)?; + validate_table_entry_version_and_id(&intent.destination)?; + if intent.source.table_bucket != intent.table_bucket + || intent.destination.table_bucket != intent.table_bucket + || intent.source.state != TableCatalogEntryState::Active + || intent.destination.state != TableCatalogEntryState::Active + { + return Err(TableCatalogStoreError::Invalid( + "catalog table rename intent has invalid table ownership or state".to_string(), + )); + } + let mut expected_destination = intent.source.clone(); + expected_destination.namespace.clone_from(&intent.destination.namespace); + expected_destination.table.clone_from(&intent.destination.table); + expected_destination.updated_at.clone_from(&intent.destination.updated_at); + if expected_destination != intent.destination + || intent.destination.updated_at.as_deref() != Some(intent.created_at.as_str()) + || (intent.source.namespace == intent.destination.namespace && intent.source.table == intent.destination.table) + { + return Err(TableCatalogStoreError::Invalid( + "catalog table rename intent changes fields other than the table identifier and update time".to_string(), + )); + } + Ok(()) +} + +fn next_table_catalog_update_time(previous: Option<&str>) -> String { + let now = OffsetDateTime::now_utc(); + let update_time = previous + .and_then(|value| OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339).ok()) + .filter(|previous| *previous >= now) + .map_or(now, |previous| previous.saturating_add(Duration::nanoseconds(1))); + update_time + .format(&time::format_description::well_known::Rfc3339) + .unwrap_or_else(|_| update_time.to_string()) +} + pub(super) fn validate_view_entry_object( paths: &TableCatalogObjectPaths, object: &str, @@ -670,6 +729,437 @@ where self.backend.put_object_unlocked(bucket, object, data, precondition).await } + async fn write_exact_entry_unlocked( + &self, + bucket: &str, + object: &str, + entry: &T, + precondition: TableCatalogPutPrecondition, + ) -> TableCatalogStoreResult + where + T: DeserializeOwned + PartialEq + Serialize, + { + let write_result = self.write_entry_unlocked(bucket, object, entry, precondition).await; + let current = self.read_entry_unlocked::(bucket, object).await?; + match current { + Some((current, Some(etag))) if current == *entry => Ok(etag), + Some((current, None)) if current == *entry => Err(TableCatalogStoreError::Internal(format!( + "catalog entry has no etag after write: {object}" + ))), + _ => match write_result { + Ok(()) => Err(TableCatalogStoreError::Internal(format!( + "catalog entry does not match the completed write: {object}" + ))), + Err(err) => Err(err), + }, + } + } + + async fn read_table_bucket_with_etag_unlocked( + &self, + table_bucket: &str, + ) -> TableCatalogStoreResult> { + let object = self.paths.table_bucket_entry_path(table_bucket); + let Some((entry, etag)) = self + .read_entry_unlocked::(self.catalog_bucket(), &object) + .await? + else { + return Ok(None); + }; + validate_table_bucket_entry_object(&self.paths, &object, &entry)?; + let Some(etag) = etag else { + return Err(TableCatalogStoreError::Internal(format!( + "catalog table bucket entry has no etag: {object}" + ))); + }; + Ok(Some((entry, etag))) + } + + async fn ensure_no_active_table_rename(&self, table_bucket: &str) -> TableCatalogStoreResult<()> { + self.table_rename_read_version(table_bucket).await.map(|_| ()) + } + + async fn table_rename_read_version(&self, table_bucket: &str) -> TableCatalogStoreResult> { + self.table_rename_read_snapshot(table_bucket) + .await + .map(|snapshot| snapshot.map(|(_, etag)| etag)) + } + + async fn table_rename_read_snapshot( + &self, + table_bucket: &str, + ) -> TableCatalogStoreResult> { + let object = self.paths.table_bucket_entry_path(table_bucket); + let Some((entry, etag)) = self.read_entry::(self.catalog_bucket(), &object).await? else { + return Ok(None); + }; + validate_table_bucket_entry_object(&self.paths, &object, &entry)?; + if let Some(rename_id) = entry.active_rename_id { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {table_bucket} has an active table rename {rename_id}" + ))); + } + let etag = + etag.ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog table bucket entry has no etag: {object}")))?; + Ok(Some((entry, etag))) + } + + async fn finish_table_rename_read(&self, table_bucket: &str, expected_version: Option<&str>) -> TableCatalogStoreResult<()> { + let object = self.paths.table_bucket_entry_path(table_bucket); + let current_version = + match self.backend.object_metadata(self.catalog_bucket(), &object).await? { + None => None, + Some(metadata) => Some(metadata.etag.ok_or_else(|| { + TableCatalogStoreError::Internal(format!("catalog table bucket entry has no etag: {object}")) + })?), + }; + if current_version.as_deref() != expected_version { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {table_bucket} changed while reading the table catalog" + ))); + } + Ok(()) + } + + async fn acquire_catalog_write_locks(&self, mut objects: Vec) -> TableCatalogStoreResult> { + objects.sort_unstable(); + objects.dedup(); + let mut guards = Vec::with_capacity(objects.len()); + for object in objects { + guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?); + } + Ok(guards) + } + + async fn advance_table_rename_intent_unlocked( + &self, + object: &str, + intent: &mut TableRenameIntent, + etag: String, + state: TableRenameIntentState, + ) -> TableCatalogStoreResult { + if intent.state >= state { + return Ok(etag); + } + intent.state = state; + intent.updated_at = OffsetDateTime::now_utc().to_string(); + self.write_exact_entry_unlocked(self.catalog_bucket(), object, intent, TableCatalogPutPrecondition::IfMatch(etag)) + .await + } + + async fn recover_active_table_rename( + &self, + table_bucket: &str, + publication: &(dyn TableCommitPublication + Sync), + ) -> TableCatalogStoreResult<()> { + if !publication.holds_table_bucket(table_bucket) { + return Err(TableCatalogStoreError::Internal( + "table rename recovery requires a table-bucket publication fence".to_string(), + )); + } + let bucket_object = self.paths.table_bucket_entry_path(table_bucket); + let Some((observed_bucket, _)) = self + .read_entry::(self.catalog_bucket(), &bucket_object) + .await? + else { + // Recovery only owns an active rename advertised by the bucket + // entry. Callers retain their existing validation when no entry + // exists. + return Ok(()); + }; + validate_table_bucket_entry_object(&self.paths, &bucket_object, &observed_bucket)?; + if observed_bucket.active_rename_id.is_none() { + return Ok(()); + } + let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_object).await?; + let Some((bucket_entry, _)) = self.read_table_bucket_with_etag_unlocked(table_bucket).await? else { + return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); + }; + let Some(rename_id) = bucket_entry.active_rename_id.clone() else { + return Ok(()); + }; + let intent_object = self.paths.table_rename_intent_path(table_bucket, &rename_id); + let _intent_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &intent_object).await?; + let Some((mut intent, mut intent_etag)) = self + .read_entry_unlocked::(self.catalog_bucket(), &intent_object) + .await? + else { + return Err(TableCatalogStoreError::Unavailable(format!( + "active table rename is missing its durable intent: {intent_object}" + ))); + }; + let Some(mut intent_etag) = intent_etag.take() else { + return Err(TableCatalogStoreError::Internal(format!( + "catalog table rename intent has no etag: {intent_object}" + ))); + }; + validate_table_rename_intent_object(&self.paths, &intent_object, &intent)?; + if intent.table_bucket != table_bucket || intent.rename_id != rename_id { + return Err(TableCatalogStoreError::Invalid( + "active table rename does not belong to its table bucket fence".to_string(), + )); + } + if bucket_entry.state != TableCatalogEntryState::Active { + return Err(TableCatalogStoreError::Conflict(format!( + "table bucket {table_bucket} became inactive during table rename recovery" + ))); + } + + let source_namespace = parse_namespace_for_store(&intent.source.namespace)?; + let source_table = parse_table_for_store(&intent.source.table)?; + let destination_namespace = parse_namespace_for_store(&intent.destination.namespace)?; + let destination_table = parse_table_for_store(&intent.destination.table)?; + let source_object = self.paths.table_entry_path(table_bucket, &source_namespace, &source_table); + let destination_object = self + .paths + .table_entry_path(table_bucket, &destination_namespace, &destination_table); + let destination_view_object = self + .paths + .view_entry_path(table_bucket, &destination_namespace, &destination_table); + let index = table_warehouse_index_entry(&intent.source)?; + let index_object = self + .paths + .warehouse_index_entry_path(table_bucket, &index.warehouse_object_prefix); + let _catalog_guards = self + .acquire_catalog_write_locks(vec![ + self.paths.namespace_entry_path(table_bucket, &source_namespace), + self.paths.namespace_entry_path(table_bucket, &destination_namespace), + source_object.clone(), + destination_object.clone(), + destination_view_object.clone(), + index_object.clone(), + ]) + .await?; + if !publication.holds_table_bucket(table_bucket) { + return Err(TableCatalogStoreError::Unavailable( + "table-bucket publication fence was lost during table rename recovery".to_string(), + )); + } + self.require_active_namespace_unlocked( + table_bucket, + &source_namespace, + &self.paths.namespace_entry_path(table_bucket, &source_namespace), + ) + .await + .map_err(|err| match err { + TableCatalogStoreError::NotFound(_) => { + TableCatalogStoreError::Conflict("table rename source namespace disappeared during recovery".to_string()) + } + err => err, + })?; + self.require_active_namespace_unlocked( + table_bucket, + &destination_namespace, + &self.paths.namespace_entry_path(table_bucket, &destination_namespace), + ) + .await + .map_err(|err| match err { + TableCatalogStoreError::NotFound(_) => { + TableCatalogStoreError::Conflict("table rename destination namespace disappeared during recovery".to_string()) + } + err => err, + })?; + if self + .read_entry_unlocked::(self.catalog_bucket(), &destination_view_object) + .await? + .is_some() + { + return Err(TableCatalogStoreError::Conflict( + "table rename destination became a view during recovery".to_string(), + )); + } + + let mut source_fence = intent.source.clone(); + source_fence.state = TableCatalogEntryState::Renaming; + // Keep the source object as a conditional-replacement tombstone instead of relying on an unconditional delete. + let mut source_tombstone = intent.source.clone(); + source_tombstone.state = TableCatalogEntryState::Deleted; + source_tombstone.updated_at = Some(intent.created_at.clone()); + match self + .read_table_with_etag_unlocked(table_bucket, &source_namespace, &source_table) + .await? + { + Some((current, _)) if current == source_fence || current == source_tombstone => {} + Some((current, current_etag)) if current == intent.source && current_etag == intent.source_etag => { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &source_object, + &source_fence, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await?; + } + _ => { + return Err(TableCatalogStoreError::Conflict(format!( + "table rename source changed during recovery: {table_bucket}/{}/{}", + source_namespace.public_name(), + source_table.as_str() + ))); + } + } + intent_etag = self + .advance_table_rename_intent_unlocked(&intent_object, &mut intent, intent_etag, TableRenameIntentState::SourceFenced) + .await?; + + let mut destination_fence = intent.destination.clone(); + destination_fence.state = TableCatalogEntryState::Renaming; + match self + .read_table_with_etag_unlocked(table_bucket, &destination_namespace, &destination_table) + .await? + { + Some((current, _)) if current == destination_fence || current == intent.destination => {} + None if intent.destination_etag.is_none() => { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &destination_object, + &destination_fence, + TableCatalogPutPrecondition::IfAbsent, + ) + .await?; + } + Some((current, current_etag)) + if current.state == TableCatalogEntryState::Deleted + && Some(current_etag.as_str()) == intent.destination_etag.as_deref() => + { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &destination_object, + &destination_fence, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await?; + } + _ => { + return Err(TableCatalogStoreError::Conflict(format!( + "table rename destination changed during recovery: {table_bucket}/{}/{}", + destination_namespace.public_name(), + destination_table.as_str() + ))); + } + } + intent_etag = self + .advance_table_rename_intent_unlocked( + &intent_object, + &mut intent, + intent_etag, + TableRenameIntentState::DestinationWritten, + ) + .await?; + + match self + .read_table_with_etag_unlocked(table_bucket, &source_namespace, &source_table) + .await? + { + Some((current, _)) if current == source_tombstone => {} + Some((current, current_etag)) if current == source_fence => { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &source_object, + &source_tombstone, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await?; + } + _ => { + return Err(TableCatalogStoreError::Conflict(format!( + "table rename source changed during recovery: {table_bucket}/{}/{}", + source_namespace.public_name(), + source_table.as_str() + ))); + } + } + intent_etag = self + .advance_table_rename_intent_unlocked( + &intent_object, + &mut intent, + intent_etag, + TableRenameIntentState::SourceTombstoned, + ) + .await?; + + let destination_index = table_warehouse_index_entry(&intent.destination)?; + let Some((current_index, current_index_etag)) = self + .read_entry_unlocked::(self.catalog_bucket(), &index_object) + .await? + else { + return Err(TableCatalogStoreError::Conflict( + "table rename warehouse index disappeared during recovery".to_string(), + )); + }; + validate_table_warehouse_index_entry_object(&self.paths, &index_object, ¤t_index)?; + if current_index != destination_index { + if current_index != index || current_index_etag.as_deref() != Some(intent.warehouse_index_etag.as_str()) { + return Err(TableCatalogStoreError::Conflict( + "table rename warehouse index changed during recovery".to_string(), + )); + } + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &index_object, + &destination_index, + TableCatalogPutPrecondition::IfMatch(intent.warehouse_index_etag.clone()), + ) + .await?; + } + intent_etag = self + .advance_table_rename_intent_unlocked( + &intent_object, + &mut intent, + intent_etag, + TableRenameIntentState::IndexPublished, + ) + .await?; + match self + .read_table_with_etag_unlocked(table_bucket, &destination_namespace, &destination_table) + .await? + { + Some((current, _)) if current == intent.destination => {} + Some((current, current_etag)) if current == destination_fence => { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &destination_object, + &intent.destination, + TableCatalogPutPrecondition::IfMatch(current_etag), + ) + .await?; + } + _ => { + return Err(TableCatalogStoreError::Conflict(format!( + "table rename destination changed during recovery: {table_bucket}/{}/{}", + destination_namespace.public_name(), + destination_table.as_str() + ))); + } + } + intent_etag = self + .advance_table_rename_intent_unlocked( + &intent_object, + &mut intent, + intent_etag, + TableRenameIntentState::DestinationPublished, + ) + .await?; + self.advance_table_rename_intent_unlocked(&intent_object, &mut intent, intent_etag, TableRenameIntentState::Completed) + .await?; + + let Some((mut bucket_entry, bucket_etag)) = self.read_table_bucket_with_etag_unlocked(table_bucket).await? else { + return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); + }; + if bucket_entry.active_rename_id.as_deref() != Some(rename_id.as_str()) { + return Err(TableCatalogStoreError::Conflict("table rename fence changed during recovery".to_string())); + } + bucket_entry.active_rename_id = None; + bucket_entry.updated_at = Some(next_table_catalog_update_time(bucket_entry.updated_at.as_deref())); + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &bucket_object, + &bucket_entry, + TableCatalogPutPrecondition::IfMatch(bucket_etag), + ) + .await?; + Ok(()) + } + async fn write_warehouse_index_state_unlocked(&self, table_bucket: &str) -> TableCatalogStoreResult<()> { let state = TableWarehouseIndexStateEntry { version: TABLE_WAREHOUSE_INDEX_STATE_VERSION, @@ -796,6 +1286,9 @@ where let candidate = table_warehouse_index_entry(entry)?; validate_table_entry_version_and_id(entry)?; for existing in self.list_all_table_entries(&candidate.table_bucket).await? { + if existing.state != TableCatalogEntryState::Active { + continue; + } if existing.table_id == candidate.table_id { if existing.namespace != candidate.namespace || existing.table != candidate.table { return Err(TableCatalogStoreError::Conflict( @@ -804,9 +1297,6 @@ where } continue; } - if existing.state != TableCatalogEntryState::Active { - continue; - } let existing_prefix = table_warehouse_object_prefix(&existing)?; if warehouse_object_prefixes_overlap(&existing_prefix, &candidate.warehouse_object_prefix) { return Err(TableCatalogStoreError::Conflict(format!( @@ -1141,7 +1631,12 @@ where if self.read_warehouse_index_state_unlocked(table_bucket).await? { return Ok(()); } - let tables = self.list_all_table_entries(table_bucket).await?; + let tables = self + .list_all_table_entries(table_bucket) + .await? + .into_iter() + .filter(|table| table.state == TableCatalogEntryState::Active) + .collect::>(); let mut table_ids = BTreeSet::new(); if let Some(table) = tables.iter().find(|table| !table_ids.insert(table.table_id.as_str())) { return Err(TableCatalogStoreError::Conflict(format!( @@ -1150,7 +1645,7 @@ where ))); } let mut active_prefixes = Vec::new(); - for table in tables.iter().filter(|table| table.state == TableCatalogEntryState::Active) { + for table in &tables { active_prefixes.push((table_warehouse_object_prefix(table)?, table.table_id.as_str())); } active_prefixes.sort_unstable_by(|left, right| left.0.cmp(&right.0)); @@ -1164,9 +1659,6 @@ where ))); } for table in tables { - if table.state != TableCatalogEntryState::Active { - continue; - } self.backfill_active_table_warehouse_index(&table.table_bucket, &table.namespace, &table.table) .await?; } @@ -1307,6 +1799,7 @@ where let _publication_completion = TableCommitPublicationCompletion::new(publication); self.require_table_bucket(&entry.table_bucket).await?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?; + self.recover_active_table_rename(&entry.table_bucket, publication).await?; let namespace_path = self.paths.namespace_entry_path(&entry.table_bucket, &namespace); let _namespace_guard = self .backend @@ -1327,6 +1820,23 @@ where entry.table_bucket, entry.namespace, entry.table ))); } + let mut precondition = precondition; + if matches!(precondition, TableCatalogPutPrecondition::IfAbsent) + && let Some((current, etag)) = self + .read_entry_unlocked::(self.catalog_bucket(), &table_path) + .await? + { + validate_table_entry_object(&self.paths, &table_path, ¤t)?; + if current.state != TableCatalogEntryState::Deleted { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: table {}/{}/{}", + entry.table_bucket, entry.namespace, entry.table + ))); + } + precondition = etag + .map(TableCatalogPutPrecondition::IfMatch) + .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog table entry has no etag: {table_path}")))?; + } // Preserve catalog -> publication -> object lock order across rolling upgrades. publication .prepare(&entry.table_bucket, &entry.namespace, &entry.table) @@ -1382,6 +1892,7 @@ where let view = parse_table_for_store(&entry.view)?; validate_view_warehouse_location(&entry.table_bucket, &entry.warehouse_location)?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?; + self.recover_active_table_rename(&entry.table_bucket, publication).await?; let namespace_path = self.paths.namespace_entry_path(&entry.table_bucket, &namespace); let _namespace_guard = self .backend @@ -1620,7 +2131,11 @@ where ) -> TableCatalogStoreResult { let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; + let publication = TableCommitLockPublication::new(&self.backend); + publication.begin_table_bucket(table_bucket).await?; + let _publication_completion = TableCommitPublicationCompletion::new(&publication); let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; + self.recover_active_table_rename(table_bucket, &publication).await?; let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; let Some((entry, _)) = self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await? else { @@ -4006,12 +4521,21 @@ where Ok(Some(entry)) } - async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> { + async fn put_table_bucket(&self, mut entry: TableBucketEntry) -> TableCatalogStoreResult<()> { validate_table_bucket_entry(&entry)?; let _registry_guard = self.acquire_table_bucket_registry_write_permit().await?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?; let object = self.paths.table_bucket_entry_path(&entry.table_bucket); let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?; + if let Some((current, _)) = self.read_table_bucket_with_etag_unlocked(&entry.table_bucket).await? { + if current.active_rename_id.is_some() { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {} has an active table rename", + entry.table_bucket + ))); + } + entry.updated_at = Some(next_table_catalog_update_time(current.updated_at.as_deref())); + } self.write_entry_unlocked(self.catalog_bucket(), &object, &entry, TableCatalogPutPrecondition::Any) .await } @@ -4023,6 +4547,16 @@ where let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?; let bucket_path = self.paths.table_bucket_entry_path(&entry.table_bucket); let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_path).await?; + if self + .read_table_bucket_with_etag_unlocked(&entry.table_bucket) + .await? + .is_some_and(|(current, _)| current.active_rename_id.is_some()) + { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {} has an active table rename", + entry.table_bucket + ))); + } let object = self.paths.namespace_entry_path(&entry.table_bucket, &namespace); let _namespace_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?; let precondition = match self @@ -4150,6 +4684,15 @@ where let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; let bucket_path = self.paths.table_bucket_entry_path(table_bucket); let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_path).await?; + if self + .read_table_bucket_with_etag_unlocked(table_bucket) + .await? + .is_some_and(|(current, _)| current.active_rename_id.is_some()) + { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {table_bucket} has an active table rename" + ))); + } let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace); // Match create_namespace and migration lock order while draining table/view creation. let _namespace_guard = self @@ -4209,6 +4752,7 @@ where } async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + let read_version = self.table_rename_read_version(table_bucket).await?; let namespace = parse_namespace_for_store(namespace)?; let mut entries = Vec::new(); for object in self @@ -4228,16 +4772,20 @@ where } } entries.sort_by(|left, right| left.table.cmp(&right.table)); + self.finish_table_rename_read(table_bucket, read_version.as_deref()).await?; Ok(entries) } async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult> { - self.list_all_table_entries(table_bucket).await.map(|entries| { + let read_version = self.table_rename_read_version(table_bucket).await?; + let entries = self.list_all_table_entries(table_bucket).await.map(|entries| { entries .into_iter() .filter(|entry| entry.state == TableCatalogEntryState::Active) .collect() - }) + })?; + self.finish_table_rename_read(table_bucket, read_version.as_deref()).await?; + Ok(entries) } async fn list_tables_page( @@ -4247,22 +4795,236 @@ where cursor: Option<&str>, limit: NonZeroUsize, ) -> TableCatalogStoreResult> { + let read_version = self.table_rename_read_version(table_bucket).await?; let namespace = parse_namespace_for_store(namespace)?; - self.list_entry_page( - &self.paths.table_entries_prefix(table_bucket, &namespace), - TABLE_ENTRY_FILE, - cursor, - limit, - |entry: &TableEntry| entry.state == TableCatalogEntryState::Active, - |object, entry: &TableEntry| validate_table_entry_object(&self.paths, object, entry).map(|_| ()), - ) - .await + let page = self + .list_entry_page( + &self.paths.table_entries_prefix(table_bucket, &namespace), + TABLE_ENTRY_FILE, + cursor, + limit, + |entry: &TableEntry| entry.state == TableCatalogEntryState::Active, + |object, entry: &TableEntry| validate_table_entry_object(&self.paths, object, entry).map(|_| ()), + ) + .await?; + self.finish_table_rename_read(table_bucket, read_version.as_deref()).await?; + Ok(page) } async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult> { - self.load_table_entry(table_bucket, namespace, table) + let read_version = self.table_rename_read_version(table_bucket).await?; + let entry = self + .load_table_entry(table_bucket, namespace, table) .await - .map(|entry| entry.filter(|table| table.state == TableCatalogEntryState::Active)) + .map(|entry| entry.filter(|table| table.state == TableCatalogEntryState::Active))?; + self.finish_table_rename_read(table_bucket, read_version.as_deref()).await?; + Ok(entry) + } + + async fn rename_table( + &self, + table_bucket: &str, + source_namespace: &str, + source_table: &str, + destination_namespace: &str, + destination_table: &str, + ) -> TableCatalogStoreResult<()> { + let source_namespace = parse_namespace_for_store(source_namespace)?; + let source_table = parse_table_for_store(source_table)?; + let destination_namespace = parse_namespace_for_store(destination_namespace)?; + let destination_table = parse_table_for_store(destination_table)?; + let publication = TableCommitLockPublication::new(&self.backend); + publication.begin_table_bucket(table_bucket).await?; + if !publication.holds_table_bucket(table_bucket) { + return Err(TableCatalogStoreError::Internal( + "table rename requires a table-bucket publication fence".to_string(), + )); + } + let _publication_completion = TableCommitPublicationCompletion::new(&publication); + let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; + self.recover_active_table_rename(table_bucket, &publication).await?; + + { + let bucket_object = self.paths.table_bucket_entry_path(table_bucket); + let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_object).await?; + let Some((mut bucket_entry, bucket_etag)) = self.read_table_bucket_with_etag_unlocked(table_bucket).await? else { + return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); + }; + if bucket_entry.state != TableCatalogEntryState::Active { + return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); + } + if bucket_entry.active_rename_id.is_some() { + return Err(TableCatalogStoreError::Unavailable(format!( + "table bucket {table_bucket} has an active table rename" + ))); + } + + let rename_id = Uuid::new_v4().to_string(); + let intent_object = self.paths.table_rename_intent_path(table_bucket, &rename_id); + let _intent_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &intent_object).await?; + let source_object = self.paths.table_entry_path(table_bucket, &source_namespace, &source_table); + let destination_object = self + .paths + .table_entry_path(table_bucket, &destination_namespace, &destination_table); + let _catalog_guards = self + .acquire_catalog_write_locks(vec![ + self.paths.namespace_entry_path(table_bucket, &source_namespace), + self.paths.namespace_entry_path(table_bucket, &destination_namespace), + source_object.clone(), + destination_object.clone(), + self.paths + .view_entry_path(table_bucket, &destination_namespace, &destination_table), + ]) + .await?; + self.require_active_namespace_unlocked( + table_bucket, + &source_namespace, + &self.paths.namespace_entry_path(table_bucket, &source_namespace), + ) + .await + .map_err(|err| match err { + TableCatalogStoreError::NotFound(_) => TableCatalogStoreError::TableNotFound(format!( + "{table_bucket}/{}/{}", + source_namespace.public_name(), + source_table.as_str() + )), + err => err, + })?; + self.require_active_namespace_unlocked( + table_bucket, + &destination_namespace, + &self.paths.namespace_entry_path(table_bucket, &destination_namespace), + ) + .await + .map_err(|err| match err { + TableCatalogStoreError::NotFound(_) => { + TableCatalogStoreError::NamespaceNotFound(format!("{table_bucket}/{}", destination_namespace.public_name())) + } + err => err, + })?; + let Some((source, source_etag)) = self + .read_table_with_etag_unlocked(table_bucket, &source_namespace, &source_table) + .await? + else { + return Err(TableCatalogStoreError::TableNotFound(format!( + "{table_bucket}/{}/{}", + source_namespace.public_name(), + source_table.as_str() + ))); + }; + if source.state != TableCatalogEntryState::Active { + return Err(TableCatalogStoreError::TableNotFound(format!( + "{table_bucket}/{}/{}", + source_namespace.public_name(), + source_table.as_str() + ))); + } + if !is_valid_table_metadata_location_for_entry(&source, &source.metadata_location) { + return Err(TableCatalogStoreError::Invalid( + "current metadata location must be inside the table metadata directory".to_string(), + )); + } + let destination_etag = match self + .read_table_with_etag_unlocked(table_bucket, &destination_namespace, &destination_table) + .await? + { + None => None, + Some((current, etag)) if current.state == TableCatalogEntryState::Deleted => Some(etag), + Some(_) => { + return Err(TableCatalogStoreError::AlreadyExists(format!( + "destination table already exists: {table_bucket}/{}/{}", + destination_namespace.public_name(), + destination_table.as_str() + ))); + } + }; + if self + .read_entry_unlocked::( + self.catalog_bucket(), + &self + .paths + .view_entry_path(table_bucket, &destination_namespace, &destination_table), + ) + .await? + .is_some() + { + return Err(TableCatalogStoreError::AlreadyExists(format!( + "destination table already exists: {table_bucket}/{}/{}", + destination_namespace.public_name(), + destination_table.as_str() + ))); + } + + let now = next_table_catalog_update_time(bucket_entry.updated_at.as_deref()); + let mut destination = source.clone(); + destination.namespace = destination_namespace.public_name(); + destination.table = destination_table.as_str().to_string(); + destination.updated_at = Some(now.clone()); + let source_index = table_warehouse_index_entry(&source)?; + let index_object = self + .paths + .warehouse_index_entry_path(table_bucket, &source_index.warehouse_object_prefix); + let _index_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &index_object).await?; + let warehouse_index_etag = match self + .read_entry_unlocked::(self.catalog_bucket(), &index_object) + .await? + { + Some((current, Some(etag))) if current == source_index => etag, + Some((current, None)) if current == source_index => { + return Err(TableCatalogStoreError::Internal(format!( + "catalog warehouse index entry has no etag: {index_object}" + ))); + } + Some(_) => { + return Err(TableCatalogStoreError::Conflict( + "table warehouse index does not match the rename source".to_string(), + )); + } + None => { + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &index_object, + &source_index, + TableCatalogPutPrecondition::IfAbsent, + ) + .await? + } + }; + let intent = TableRenameIntent { + version: TABLE_RENAME_INTENT_VERSION, + rename_id: rename_id.clone(), + table_bucket: table_bucket.to_string(), + source, + destination, + source_etag, + destination_etag, + warehouse_index_etag, + state: TableRenameIntentState::Prepared, + created_at: now.clone(), + updated_at: now.clone(), + }; + validate_table_rename_intent_object(&self.paths, &intent_object, &intent)?; + + // An orphan intent has no catalog effect; a published bucket fence without its intent cannot be recovered safely. + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &intent_object, + &intent, + TableCatalogPutPrecondition::IfAbsent, + ) + .await?; + bucket_entry.active_rename_id = Some(rename_id); + bucket_entry.updated_at = Some(now); + self.write_exact_entry_unlocked( + self.catalog_bucket(), + &bucket_object, + &bucket_entry, + TableCatalogPutPrecondition::IfMatch(bucket_etag), + ) + .await?; + } + + self.recover_active_table_rename(table_bucket, &publication).await } async fn resolve_table_data_plane_resource( @@ -4273,7 +5035,7 @@ where if table_bucket.is_empty() || object.is_empty() { return Ok(None); } - let Some(table_bucket_entry) = self.get_table_bucket(table_bucket).await? else { + let Some((table_bucket_entry, read_version)) = self.table_rename_read_snapshot(table_bucket).await? else { return Err(TableCatalogStoreError::Internal(format!( "object-backed catalog has no entry for table-enabled bucket {table_bucket}" ))); @@ -4284,34 +5046,36 @@ where ))); } - if self.warehouse_index_ready(table_bucket).await? { - return match self + 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, - }; - } - - 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, - }, - 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 } - Err(err) => Err(err), - } + } 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, + }, + 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 + } + Err(err) => Err(err), + } + }?; + self.finish_table_rename_read(table_bucket, Some(&read_version)).await?; + Ok(resource) } async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult { @@ -4330,6 +5094,11 @@ where let namespace = parse_namespace_for_store(&request.namespace)?; let table = parse_table_for_store(&request.table)?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(&request.table_bucket).await?; + if publication.holds_table_bucket(&request.table_bucket) { + self.recover_active_table_rename(&request.table_bucket, publication).await?; + } else { + self.ensure_no_active_table_rename(&request.table_bucket).await?; + } let table_path = self.paths.table_entry_path(&request.table_bucket, &namespace, &table); let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; // Preserve catalog -> publication -> object lock order across rolling upgrades. @@ -4749,6 +5518,7 @@ where let namespace = parse_namespace_for_store(namespace)?; let table = parse_table_for_store(table)?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; + self.recover_active_table_rename(table_bucket, &publication).await?; let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace); let _namespace_guard = self .backend @@ -4902,6 +5672,11 @@ where } } let _migration_guard = self.acquire_object_backed_catalog_write_permit(&request.table_bucket).await?; + if publication.holds_table_bucket(&request.table_bucket) { + self.recover_active_table_rename(&request.table_bucket, publication).await?; + } else { + self.ensure_no_active_table_rename(&request.table_bucket).await?; + } let namespace_path = self.paths.namespace_entry_path(&request.table_bucket, &namespace); let _namespace_guard = self .backend @@ -5020,9 +5795,13 @@ where } async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()> { + let publication = TableCommitLockPublication::new(&self.backend); + publication.begin_table_bucket(table_bucket).await?; + let _publication_completion = TableCommitPublicationCompletion::new(&publication); let namespace = parse_namespace_for_store(namespace)?; let view = parse_table_for_store(view)?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(table_bucket).await?; + self.recover_active_table_rename(table_bucket, &publication).await?; let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace); let _namespace_guard = self .backend diff --git a/rustfs/src/table_catalog/test_support.rs b/rustfs/src/table_catalog/test_support.rs index 1daeeb7a0..5967e91e6 100644 --- a/rustfs/src/table_catalog/test_support.rs +++ b/rustfs/src/table_catalog/test_support.rs @@ -642,12 +642,26 @@ impl TestCatalogObjectBackend { let mut state = self.state.lock().await; let key = (bucket.to_string(), object.to_string()); let next_attempt = state.put_attempts.get(&key).copied().unwrap_or_default() + 1; + Self::pause_put_attempt_unlocked(&mut state, key, next_attempt) + } + + pub(crate) async fn pause_put_attempt(&self, bucket: &str, object: &str, attempt: usize) -> TestCatalogObjectPause { + let mut state = self.state.lock().await; + let key = (bucket.to_string(), object.to_string()); + Self::pause_put_attempt_unlocked(&mut state, key, attempt) + } + + fn pause_put_attempt_unlocked( + state: &mut TestCatalogObjectState, + key: (String, String), + attempt: usize, + ) -> TestCatalogObjectPause { let pause = TestCatalogObjectPause::default(); state .pause_put_attempts .entry(key) .or_default() - .insert(next_attempt, pause.clone()); + .insert(attempt, pause.clone()); pause } diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 8c1363407..5e88cc213 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -144,6 +144,7 @@ fn catalog_entry_structures_serialize_stable_fields() { warehouse_root: "s3://analytics/".to_string(), state: TableCatalogEntryState::Active, properties: BTreeMap::from([("owner".to_string(), "platform".to_string())]), + active_rename_id: None, created_at: Some("2026-05-23T00:00:00Z".to_string()), updated_at: Some("2026-05-23T00:00:00Z".to_string()), }; @@ -3441,6 +3442,7 @@ fn test_bucket_entry(bucket: &str) -> TableBucketEntry { warehouse_root: format!("s3://{bucket}/"), state: TableCatalogEntryState::Active, properties: BTreeMap::new(), + active_rename_id: None, created_at: None, updated_at: None, } @@ -5107,7 +5109,8 @@ async fn object_catalog_pagination_bounds_reads_and_covers_rest_resources() { .await .expect("first table page should load"); assert_eq!(table_page.entries[0].table, "alpha"); - assert_eq!(backend.read_call_count().await, 1); + // One read snapshots the bucket rename fence and one loads the page entry. + assert_eq!(backend.read_call_count().await, 2); let table_page = store .list_tables_page(bucket, &namespace_name, table_page.next_cursor.as_deref(), one) .await @@ -17752,7 +17755,461 @@ async fn strong_catalog_table_rename_returns_success_after_committed_snapshot_re } #[tokio::test] -async fn configured_object_catalog_rejects_table_rename() { +async fn object_catalog_table_rename_preserves_identity_index_and_reuses_source_tombstone() { + let backend = TestCatalogObjectBackend { + content_addressed_etags: true, + ..Default::default() + }; + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let source_namespace = Namespace::parse("sales").expect("source namespace should parse"); + let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse"); + let source_table = IdentifierSegment::parse("orders").expect("source table should parse"); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &source_namespace)) + .await + .unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &destination_namespace)) + .await + .unwrap(); + let source = test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"), + ); + store.create_table(source.clone()).await.unwrap(); + let bucket_object = store.paths.table_bucket_entry_path(bucket); + let bucket_etag_before = store + .read_entry::(RUSTFS_META_BUCKET, &bucket_object) + .await + .unwrap() + .unwrap() + .1 + .expect("table bucket should have an etag"); + + store + .rename_table(bucket, "sales", "orders", "curated", "orders_v2") + .await + .expect("object-backed table rename should complete"); + let bucket_etag_after = store + .read_entry::(RUSTFS_META_BUCKET, &bucket_object) + .await + .unwrap() + .unwrap() + .1 + .expect("table bucket should have an etag"); + assert_ne!(bucket_etag_after, bucket_etag_before); + + assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_none()); + let destination = store + .load_table(bucket, "curated", "orders_v2") + .await + .unwrap() + .expect("destination table should exist"); + let mut expected_destination = source.clone(); + expected_destination.namespace = "curated".to_string(); + expected_destination.table = "orders_v2".to_string(); + assert_ne!(destination.updated_at, source.updated_at); + expected_destination.updated_at.clone_from(&destination.updated_at); + assert_eq!(destination, expected_destination); + + let source_object = store.paths.table_entry_path(bucket, &source_namespace, &source_table); + let source_tombstone = store + .read_entry::(RUSTFS_META_BUCKET, &source_object) + .await + .unwrap() + .expect("source tombstone should remain") + .0; + assert_eq!(source_tombstone.state, TableCatalogEntryState::Deleted); + assert_eq!(source_tombstone.table_id, destination.table_id); + assert!( + store + .get_table_bucket(bucket) + .await + .unwrap() + .expect("table bucket should exist") + .active_rename_id + .is_none() + ); + + let destination_index = table_warehouse_index_entry(&destination).unwrap(); + let index_object = store + .paths + .warehouse_index_entry_path(bucket, &destination_index.warehouse_object_prefix); + let persisted_index = store + .read_entry::(RUSTFS_META_BUCKET, &index_object) + .await + .unwrap() + .expect("warehouse index should exist") + .0; + assert_eq!(persisted_index, destination_index); + let resource = store + .resolve_table_data_plane_resource(bucket, "tables/table-id/data/part.parquet") + .await + .unwrap() + .expect("renamed table should resolve its stable warehouse prefix"); + assert_eq!(resource.namespace, "curated"); + assert_eq!(resource.table, "orders_v2"); + store + .backfill_table_warehouse_index(bucket) + .await + .expect("retained source tombstone should not make the warehouse index ambiguous"); + let migration = store.plan_durable_strong_backing_migration(bucket).await.unwrap(); + assert_eq!(migration.table_count, 1); + assert!( + !migration + .blockers + .contains(&TableCatalogBackingMigrationBlocker::DuplicateTableIdentity) + ); + + store + .rename_table(bucket, "curated", "orders_v2", "sales", "orders") + .await + .expect("rename should conditionally replace the retained source tombstone"); + store + .rename_table(bucket, "sales", "orders", "curated", "orders_v2") + .await + .expect("rename should conditionally replace a destination tombstone"); + + let mut replacement = test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00002.metadata.json"), + ); + replacement.table_id = "replacement-table-id".to_string(); + replacement.table_uuid = "replacement-table-uuid".to_string(); + replacement.warehouse_location = "s3://analytics/tables/replacement-table-id".to_string(); + store + .create_table(replacement.clone()) + .await + .expect("create should conditionally replace the source tombstone"); + assert_eq!( + store + .load_table(bucket, "sales", "orders") + .await + .unwrap() + .expect("source identifier should be reusable"), + replacement + ); +} + +#[tokio::test] +async fn object_catalog_table_rename_rejects_missing_and_conflicting_destinations() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let source_namespace = Namespace::parse("sales").unwrap(); + let destination_namespace = Namespace::parse("curated").unwrap(); + let source_table = IdentifierSegment::parse("orders").unwrap(); + let destination_table = IdentifierSegment::parse("orders_v2").unwrap(); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &source_namespace)) + .await + .unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &destination_namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"), + )) + .await + .unwrap(); + + assert_matches!( + store.rename_table(bucket, "sales", "orders", "missing", "orders_v2").await, + Err(TableCatalogStoreError::NamespaceNotFound(_)) + ); + assert_matches!( + store.rename_table(bucket, "sales", "missing", "curated", "orders_v2").await, + Err(TableCatalogStoreError::TableNotFound(_)) + ); + let mut existing = test_table_entry( + bucket, + &destination_namespace, + &destination_table, + default_table_metadata_file_path(&destination_namespace, &destination_table, "00001.metadata.json"), + ); + existing.table_id = "destination-table-id".to_string(); + existing.table_uuid = "destination-table-uuid".to_string(); + existing.warehouse_location = "s3://analytics/tables/destination-table-id".to_string(); + store.create_table(existing).await.unwrap(); + assert_matches!( + store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await, + Err(TableCatalogStoreError::AlreadyExists(_)) + ); + let destination_view = IdentifierSegment::parse("orders_view").unwrap(); + store + .create_view(test_view_entry( + bucket, + &destination_namespace, + &destination_view, + default_view_metadata_file_path(&destination_namespace, &destination_view, "00001.metadata.json"), + )) + .await + .unwrap(); + assert_matches!( + store.rename_table(bucket, "sales", "orders", "curated", "orders_view").await, + Err(TableCatalogStoreError::AlreadyExists(_)) + ); + assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_some()); +} + +#[tokio::test] +async fn object_catalog_table_rename_fails_closed_around_durable_fence_creation() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let source_namespace = Namespace::parse("sales").unwrap(); + let destination_namespace = Namespace::parse("curated").unwrap(); + let source_table = IdentifierSegment::parse("orders").unwrap(); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &source_namespace)) + .await + .unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &destination_namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"), + )) + .await + .unwrap(); + + let bucket_object = store.paths.table_bucket_entry_path(bucket); + backend.fail_next_put(RUSTFS_META_BUCKET, &bucket_object).await; + assert_matches!( + store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await, + Err(TableCatalogStoreError::Internal(_)) + ); + assert!( + store + .get_table_bucket(bucket) + .await + .unwrap() + .expect("table bucket should remain") + .active_rename_id + .is_none() + ); + assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_some()); + assert!(store.load_table(bucket, "curated", "orders_v2").await.unwrap().is_none()); + + let mut fenced_bucket = store.get_table_bucket(bucket).await.unwrap().unwrap(); + fenced_bucket.active_rename_id = Some("missing-intent".to_string()); + backend + .seed_object( + RUSTFS_META_BUCKET, + &bucket_object, + serde_json::to_vec(&fenced_bucket).expect("fenced bucket should serialize"), + ) + .await; + assert_matches!( + store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await, + Err(TableCatalogStoreError::Unavailable(_)) + ); + assert_eq!( + store + .get_table_bucket(bucket) + .await + .unwrap() + .expect("table bucket should remain fail-closed") + .active_rename_id + .as_deref(), + Some("missing-intent") + ); + assert_matches!( + store + .resolve_table_data_plane_resource(bucket, "tables/table-id/data/part.parquet") + .await, + Err(TableCatalogStoreError::Unavailable(_)) + ); +} + +#[tokio::test] +async fn object_catalog_table_rename_recovers_after_destination_publish_and_fences_concurrent_mutations() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let source_namespace = Namespace::parse("sales").expect("source namespace should parse"); + let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse"); + let source_table = IdentifierSegment::parse("orders").expect("source table should parse"); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &source_namespace)) + .await + .unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &destination_namespace)) + .await + .unwrap(); + let source = test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"), + ); + store.create_table(source.clone()).await.unwrap(); + + let source_object = store.paths.table_entry_path(bucket, &source_namespace, &source_table); + let source_tombstone_attempt = backend.put_attempt_count(RUSTFS_META_BUCKET, &source_object).await + 2; + let source_tombstone_pause = backend + .pause_put_attempt(RUSTFS_META_BUCKET, &source_object, source_tombstone_attempt) + .await; + let rename_store = store.clone(); + let rename = tokio::spawn(async move { + rename_store + .rename_table(bucket, "sales", "orders", "curated", "orders_v2") + .await + }); + source_tombstone_pause.wait_started().await; + + let active_rename_id = store + .get_table_bucket(bucket) + .await + .unwrap() + .expect("table bucket should exist") + .active_rename_id + .expect("rename fence should be durable before destination publication"); + assert_matches!( + store.load_table(bucket, "sales", "orders").await, + Err(TableCatalogStoreError::Unavailable(_)) + ); + assert_matches!(store.list_tables(bucket, "sales").await, Err(TableCatalogStoreError::Unavailable(_))); + assert_matches!( + store + .resolve_table_data_plane_resource(bucket, "tables/table-id/data/part.parquet") + .await, + Err(TableCatalogStoreError::Unavailable(_)) + ); + let migration = store.plan_durable_strong_backing_migration(bucket).await.unwrap(); + assert_eq!(migration.status, TableCatalogBackingMigrationStatus::RecoveryRequired); + assert!( + migration + .blockers + .contains(&TableCatalogBackingMigrationBlocker::TableRenameRecoveryRequired) + ); + + let publication_lock = default_table_bucket_publication_lock_path(); + let publication_attempts = backend.write_lock_acquisition_count(bucket, &publication_lock).await; + let commit_store = store.clone(); + let commit = tokio::spawn(async move { + commit_store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: "sales".to_string(), + table: "orders".to_string(), + commit_id: "concurrent-commit".to_string(), + idempotency_key: None, + operation: "append".to_string(), + expected_version_token: source.version_token, + expected_metadata_location: source.metadata_location, + new_metadata_location: "unused.metadata.json".to_string(), + requirements: Vec::new(), + writer: Some("rename-test".to_string()), + }) + .await + }); + let drop_store = store.clone(); + let drop_table = tokio::spawn(async move { drop_store.drop_table(bucket, "sales", "orders").await }); + let create_store = store.clone(); + let mut replacement = test_table_entry( + bucket, + &source_namespace, + &source_table, + default_table_metadata_file_path(&source_namespace, &source_table, "00002.metadata.json"), + ); + replacement.table_id = "replacement-table-id".to_string(); + replacement.table_uuid = "replacement-table-uuid".to_string(); + replacement.warehouse_location = "s3://analytics/tables/replacement-table-id".to_string(); + let create = tokio::spawn(async move { create_store.create_table(replacement).await }); + tokio::time::timeout(TABLE_CATALOG_TEST_TIMEOUT, async { + while backend.write_lock_acquisition_count(bucket, &publication_lock).await < publication_attempts + 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("concurrent mutations should reach the table-bucket publication fence"); + assert!(!commit.is_finished()); + assert!(!drop_table.is_finished()); + assert!(!create.is_finished()); + commit.abort(); + drop_table.abort(); + create.abort(); + + rename.abort(); + source_tombstone_pause.release(); + let _ = rename.await; + let destination_object = store.paths.table_entry_path( + bucket, + &destination_namespace, + &IdentifierSegment::parse("orders_v2").expect("destination table should parse"), + ); + let source_fence = store + .read_entry::(RUSTFS_META_BUCKET, &source_object) + .await + .unwrap() + .expect("source rename fence should be durable") + .0; + let destination_fence = store + .read_entry::(RUSTFS_META_BUCKET, &destination_object) + .await + .unwrap() + .expect("destination rename fence should be durable") + .0; + assert_eq!(source_fence.state, TableCatalogEntryState::Renaming); + assert_eq!(destination_fence.state, TableCatalogEntryState::Renaming); + assert_matches!( + store.load_table(bucket, "curated", "orders_v2").await, + Err(TableCatalogStoreError::Unavailable(_)) + ); + assert_matches!( + store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await, + Err(TableCatalogStoreError::TableNotFound(_)) + ); + + assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_none()); + let destination = store + .load_table(bucket, "curated", "orders_v2") + .await + .unwrap() + .expect("recovery should finish the destination publication"); + assert_eq!(destination.table_id, "table-id"); + assert!( + store + .get_table_bucket(bucket) + .await + .unwrap() + .expect("table bucket should exist") + .active_rename_id + .is_none() + ); + let intent_object = store.paths.table_rename_intent_path(bucket, &active_rename_id); + let intent = store + .read_entry::(RUSTFS_META_BUCKET, &intent_object) + .await + .unwrap() + .expect("completed rename intent should be retained as a recovery record") + .0; + assert_eq!(intent.state, TableRenameIntentState::Completed); +} + +#[tokio::test] +async fn configured_object_catalog_dispatches_table_rename() { let store = ConfiguredTableCatalogStore::new_for_test(TestCatalogObjectBackend::default(), TableCatalogBackingMode::ObjectBacked); @@ -17760,6 +18217,6 @@ async fn configured_object_catalog_rejects_table_rename() { store .rename_table("analytics", "sales", "orders", "curated", "orders_v2") .await, - Err(TableCatalogStoreError::Unsupported(_)) + Err(TableCatalogStoreError::NotFound(_)) ); }