feat(table-catalog): support disabling empty catalogs (#8376)

* feat(table-catalog): support disabling empty catalogs

* fix(admin): use the storage error domain module

* style(admin): format the publication fence error

* fix(table-catalog): use shared API error constructors
This commit is contained in:
GatewayJ
2026-10-07 14:07:56 +08:00
committed by GitHub
parent 47ed6717de
commit 477f768f7e
15 changed files with 547 additions and 9 deletions
@@ -43,6 +43,24 @@ impl Operation for EnableTableBucketHandler {
}
}
pub struct DisableTableBucketHandler {}
#[async_trait::async_trait]
impl Operation for DisableTableBucketHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let resource = TableCatalogResource::warehouse(&warehouse);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?;
let backend = table_catalog_backend_from_extensions(&req.extensions)?;
let object_store = runtime_sources::object_store_from_req(&req)
.ok_or_else(|| table_catalog_internal_error("request object store is not initialized"))?;
let store = table_catalog_store_from_backend(backend.clone())?;
let publication = TableCommitObjectBackend::preauthorized(backend);
let response = disable_table_bucket_response(&store, &publication, object_store.as_ref(), &warehouse).await?;
build_json_response(StatusCode::OK, &response)
}
}
pub struct GetTableBucketHandler {}
#[async_trait::async_trait]
+84 -3
View File
@@ -181,6 +181,7 @@ const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &[];
static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {};
static DISABLE_TABLE_BUCKET_HANDLER: DisableTableBucketHandler = DisableTableBucketHandler {};
static GET_TABLE_BUCKET_HANDLER: GetTableBucketHandler = GetTableBucketHandler {};
static GET_TABLE_CATALOG_MIGRATION_HANDLER: GetTableCatalogMigrationHandler = GetTableCatalogMigrationHandler {};
static MATERIALIZE_TABLE_CATALOG_MIGRATION_HANDLER: MaterializeTableCatalogMigrationHandler =
@@ -601,6 +602,8 @@ struct ExternalCatalogBridgeSyncResponse {
#[derive(Debug, Serialize)]
struct TableBucketResponse {
#[serde(rename = "disable-supported")]
disable_supported: bool,
#[serde(rename = "table-bucket")]
table_bucket: String,
enabled: bool,
@@ -2341,11 +2344,17 @@ where
if !table_bucket_enabled {
return Err(s3_error!(InvalidRequest, "bucket {bucket} is not table-enabled"));
}
if store.get_table_bucket(bucket).await.map_err(catalog_store_error)?.is_some() {
return Ok(());
if let Some(entry) = store.get_table_bucket(bucket).await.map_err(catalog_store_error)? {
return if entry.state == crate::table_catalog::TableCatalogEntryState::Active {
Ok(())
} else {
Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::NotFound(format!(
"table bucket {bucket}"
))))
};
}
store
.put_table_bucket(table_bucket_entry_from_metadata_marker(bucket))
.ensure_table_bucket(table_bucket_entry_from_metadata_marker(bucket))
.await
.map_err(catalog_store_error)
}
@@ -2366,6 +2375,7 @@ where
};
Ok(TableBucketResponse {
disable_supported: true,
table_bucket: bucket.to_string(),
enabled,
catalog_type,
@@ -2398,10 +2408,78 @@ where
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(publication);
enable_table_bucket_marker(object_store, bucket).await?;
if let Some(mut entry) = store.get_table_bucket(bucket).await.map_err(catalog_store_error)?
&& entry.state == crate::table_catalog::TableCatalogEntryState::Deleted
{
entry.state = crate::table_catalog::TableCatalogEntryState::Active;
store.put_table_bucket(entry).await.map_err(catalog_store_error)?;
}
ensure_table_bucket_entry(store, bucket, true).await?;
table_bucket_response(store, bucket, true).await
}
async fn disable_table_bucket_response<S, B>(
store: &S,
publication: &TableCommitObjectBackend<B>,
object_store: &ECStore,
bucket: &str,
) -> S3Result<TableBucketResponse>
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
B: crate::table_catalog::TableCatalogObjectBackend,
{
crate::table_catalog::TableCommitPublication::begin_table_bucket(publication, bucket)
.await
.map_err(catalog_store_error)?;
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(publication);
if !crate::table_catalog::TableCommitPublication::holds_table_bucket(publication, bucket) {
return Err(ApiError::other("table bucket disablement requires a publication fence").into());
}
// Staged metadata can exist without a registered table. Keep its protection
// until it has been explicitly cleaned up by the catalog operator.
let reserved = publication
.backend
.list_objects_page(
bucket,
&format!("{}/", crate::table_catalog::TABLE_RESERVED_PREFIX),
None,
NonZeroUsize::MIN,
)
.await
.map_err(catalog_store_error)?;
if !reserved.objects.is_empty() || reserved.is_truncated {
return Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::Conflict(
"table bucket contains reserved metadata; catalog disablement requires an empty catalog".to_string(),
)));
}
store
.disable_empty_table_bucket(table_bucket_entry_from_metadata_marker(bucket))
.await
.map_err(catalog_store_error)?;
if !crate::table_catalog::TableCommitPublication::holds_table_bucket(publication, bucket) {
return Err(ApiError {
message: "table bucket publication fence was lost".to_string(),
..ApiError::service_unavailable()
}
.into());
}
// Persist the inactive entry before clearing the marker. A failed metadata
// update leaves S3 protections enabled and can be retried without deleting data.
object_store
.update_bucket_metadata_config_validated(bucket, crate::table_catalog::TABLE_BUCKET_MARKER_CONFIG, Vec::new(), || {
if crate::table_catalog::TableCommitPublication::holds_table_bucket(publication, bucket) {
Ok(())
} else {
Err(crate::admin::storage_api::error::StorageError::other(
"table bucket publication fence was lost",
))
}
})
.await
.map_err(|err| ApiError::other(format!("failed to disable table bucket: {err}")))?;
table_bucket_response(store, bucket, false).await
}
fn namespace_segments(namespace: &crate::table_catalog::Namespace) -> Vec<String> {
namespace
.segments()
@@ -5638,6 +5716,7 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
ensure_table_bucket_entry(store, bucket, true).await?;
let metadata_data = serde_json::to_vec(&metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize initial table metadata: {}", err))?;
metadata_backend
@@ -5678,6 +5757,7 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
ensure_table_bucket_entry(store, bucket, true).await?;
let metadata_data = serde_json::to_vec(&metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize initial view metadata: {}", err))?;
metadata_backend
@@ -6457,6 +6537,7 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
ensure_table_bucket_entry(store, bucket, true).await?;
let metadata_data = serde_json::to_vec(&metadata).map_err(|err| {
S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize initial table metadata: {err}"))
})?;
@@ -29,6 +29,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
format!("{prefix}/buckets/{{warehouse}}").as_str(),
AdminOperation(&ENABLE_TABLE_BUCKET_HANDLER),
)?;
r.insert(
Method::DELETE,
format!("{prefix}/buckets/{{warehouse}}").as_str(),
AdminOperation(&DISABLE_TABLE_BUCKET_HANDLER),
)?;
r.insert(
Method::GET,
format!("{prefix}/buckets/{{warehouse}}").as_str(),
@@ -684,6 +684,7 @@ fn table_catalog_handlers_require_table_admin_actions() {
for (handler, action) in [
("EnableTableBucketHandler", "AdminAction::SetTableBucketAction"),
("DisableTableBucketHandler", "AdminAction::SetTableBucketAction"),
("GetTableBucketHandler", "AdminAction::GetTableBucketAction"),
("GetTableCatalogMigrationHandler", "AdminAction::GetTableCatalogAction"),
("RestListNamespacesHandler", "AdminAction::GetTableNamespaceAction"),
@@ -13507,3 +13508,103 @@ async fn legacy_commit_rejects_mismatched_table_uuid_before_commit() {
assert_eq!(unchanged.metadata_location, current_location);
assert_eq!(unchanged.generation, current.generation);
}
#[tokio::test]
#[serial_test::serial]
async fn disable_table_bucket_preserves_objects_and_can_be_reenabled() {
let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
let bucket = format!("disable-{}", Uuid::new_v4().simple());
object_store
.make_bucket(&bucket, &MakeBucketOptions::default())
.await
.unwrap();
let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
object_store.clone(),
crate::table_catalog::StrongTableCatalogRuntime::default(),
);
let catalog = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
let publication = TableCommitObjectBackend::preauthorized(backend.clone());
enable_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
.await
.unwrap();
let rejected = object_store
.update_bucket_metadata_config_validated(&bucket, crate::table_catalog::TABLE_BUCKET_MARKER_CONFIG, Vec::new(), || {
Err(crate::admin::storage_api::error::StorageError::other("lost publication fence"))
})
.await;
assert!(rejected.is_err());
assert!(
object_store
.get_bucket_metadata(&bucket)
.await
.unwrap()
.table_bucket_enabled()
);
backend
.put_object(
&bucket,
"external/orders.parquet",
b"parquet-fixture".to_vec(),
crate::table_catalog::TableCatalogPutPrecondition::Any,
)
.await
.unwrap();
let publication = TableCommitObjectBackend::preauthorized(backend.clone());
let response = disable_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
.await
.unwrap();
assert!(!response.enabled);
assert!(ensure_table_bucket_entry(&catalog, &bucket, true).await.is_err());
assert!(
!object_store
.get_bucket_metadata(&bucket)
.await
.unwrap()
.table_bucket_enabled()
);
assert_eq!(
backend
.read_object(&bucket, "external/orders.parquet")
.await
.unwrap()
.unwrap()
.data,
b"parquet-fixture"
);
let publication = TableCommitObjectBackend::preauthorized(backend.clone());
assert!(
!disable_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
.await
.unwrap()
.enabled
);
let publication = TableCommitObjectBackend::preauthorized(backend.clone());
assert!(
enable_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
.await
.unwrap()
.enabled
);
backend
.put_object(
&bucket,
".rustfs-table/staged.json",
b"{}".to_vec(),
crate::table_catalog::TableCatalogPutPrecondition::Any,
)
.await
.unwrap();
let publication = TableCommitObjectBackend::preauthorized(backend);
assert!(
disable_table_bucket_response(&catalog, &publication, object_store.as_ref(), &bucket)
.await
.is_err()
);
assert!(
object_store
.get_bucket_metadata(&bucket)
.await
.unwrap()
.table_bucket_enabled()
);
}
+15 -1
View File
@@ -1019,6 +1019,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
admin(HttpMethod::Post, "/rustfs/admin/v3/oidc/validate", SERVER_INFO, RouteRiskLevel::High),
admin(HttpMethod::Get, "/iceberg/v1/config", GET_TABLE_CATALOG, RouteRiskLevel::Sensitive),
admin(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET, RouteRiskLevel::High),
admin(
HttpMethod::Delete,
"/iceberg/v1/buckets/{warehouse}",
SET_TABLE_BUCKET,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/iceberg/v1/buckets/{warehouse}",
@@ -1303,6 +1309,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
RouteRiskLevel::High,
),
admin(HttpMethod::Get, "/_iceberg/v1/config", GET_TABLE_CATALOG, RouteRiskLevel::Sensitive),
admin(
HttpMethod::Delete,
"/_iceberg/v1/buckets/{warehouse}",
SET_TABLE_BUCKET,
RouteRiskLevel::High,
),
admin(
HttpMethod::Put,
"/_iceberg/v1/buckets/{warehouse}",
@@ -1917,8 +1929,10 @@ mod tests {
let table_specs = ADMIN_ROUTE_POLICY_SPECS
.iter()
.filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
assert_eq!(table_specs.count(), 100);
assert_eq!(table_specs.count(), 102);
assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Delete, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Delete, "/_iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
@@ -457,6 +457,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
admin_route(Method::GET, "/v3/idp/openid/list-access-keys-bulk"),
table_route(Method::GET, "/config"),
table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"),
table_route_sample(Method::DELETE, "/buckets/{warehouse}", "/buckets/analytics"),
table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"),
table_route_sample(Method::GET, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
table_route_sample(Method::POST, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
@@ -659,6 +660,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
),
compat_table_route(Method::GET, "/config"),
compat_table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"),
compat_table_route_sample(Method::DELETE, "/buckets/{warehouse}", "/buckets/analytics"),
compat_table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"),
compat_table_route_sample(Method::GET, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
compat_table_route_sample(Method::POST, "/{warehouse}/catalog/migration", "/analytics/catalog/migration"),
+28
View File
@@ -139,6 +139,20 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()>;
async fn ensure_table_bucket(&self, _entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
Err(TableCatalogStoreError::Unsupported(
"table bucket initialization is not supported by this catalog store".to_string(),
))
}
/// Persists an inactive entry only when no namespace or retained resource exists.
/// The caller holds the bucket publication fence until its metadata marker is cleared.
async fn disable_empty_table_bucket(&self, _entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
Err(TableCatalogStoreError::Unsupported(
"table bucket disablement is not supported by this catalog store".to_string(),
))
}
async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()>;
async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>>;
@@ -1270,6 +1284,20 @@ where
}
}
async fn ensure_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.ensure_table_bucket(entry).await,
Self::DurableStrong(store) => store.ensure_table_bucket(entry).await,
}
}
async fn disable_empty_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.disable_empty_table_bucket(entry).await,
Self::DurableStrong(store) => store.disable_empty_table_bucket(entry).await,
}
}
async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(store) => store.create_namespace(entry).await,
+72 -4
View File
@@ -4883,6 +4883,71 @@ where
.await
}
async fn ensure_table_bucket(&self, 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? {
return if current.state == TableCatalogEntryState::Active {
Ok(())
} else {
Err(TableCatalogStoreError::NotFound(format!("table bucket {}", entry.table_bucket)))
};
}
self.write_entry_unlocked(
self.catalog_bucket(),
&object,
&entry,
TableCatalogPutPrecondition::IfAbsent,
_guard.write_commit_guards(),
)
.await
}
async fn disable_empty_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()
|| !matches!(current.state, TableCatalogEntryState::Active | TableCatalogEntryState::Deleted)
{
return Err(TableCatalogStoreError::Conflict("table bucket has an operation in progress".to_string()));
}
entry = current;
}
// Namespace creation holds this same bucket lock. Retained table, view, and
// maintenance records also prevent removing their data-plane protections.
let page = self
.backend
.list_objects_page(
self.catalog_bucket(),
&self.paths.namespace_entries_prefix(&entry.table_bucket),
None,
NonZeroUsize::MIN,
)
.await?;
if !page.objects.is_empty() || page.is_truncated {
return Err(TableCatalogStoreError::Conflict(
"table bucket contains namespaces or retained catalog resources".to_string(),
));
}
entry.state = TableCatalogEntryState::Deleted;
entry.updated_at = Some(next_table_catalog_update_time(entry.updated_at.as_deref()));
self.write_entry_unlocked(
self.catalog_bucket(),
&object,
&entry,
TableCatalogPutPrecondition::Any,
_guard.write_commit_guards(),
)
.await
}
async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {
let namespace = validate_namespace_entry_identity(&entry)?;
validate_namespace_properties(&entry.properties)?;
@@ -4890,11 +4955,14 @@ 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())
let current = self.read_table_bucket_with_etag_unlocked(&entry.table_bucket).await?;
if current
.as_ref()
.is_none_or(|(entry, _)| entry.state != TableCatalogEntryState::Active)
{
return Err(TableCatalogStoreError::NotFound(format!("table bucket {}", entry.table_bucket)));
}
if current.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
+59
View File
@@ -2046,6 +2046,65 @@ where
self.finalize_snapshot_write(snapshot, precondition, postcondition).await
}
async fn ensure_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
validate_table_bucket_entry(&entry)?;
let _migration_guard = self.acquire_snapshot_write_permit().await?;
let _write_guard = self.write_lock.lock().await;
self.hydrate_state().await?;
let (snapshot, precondition, postcondition) = {
let state = self.state.lock().await;
if state.table_buckets.contains_key(&entry.table_bucket) {
return Self::require_table_bucket_in_state(&state, &entry.table_bucket);
}
let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state);
draft_state.table_buckets.insert(entry.table_bucket.clone(), entry.clone());
(
Self::snapshot_from_mutated_state_locked(&mut draft_state, self.snapshot_write_version)?,
precondition,
StrongSnapshotWritePostcondition::TableBucketPresent(entry),
)
};
self.finalize_snapshot_write(snapshot, precondition, postcondition).await
}
async fn disable_empty_table_bucket(&self, mut entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
validate_table_bucket_entry(&entry)?;
let _migration_guard = self.acquire_snapshot_write_permit().await?;
let _write_guard = self.write_lock.lock().await;
self.hydrate_state().await?;
let (snapshot, precondition, postcondition) = {
let state = self.state.lock().await;
if let Some(current) = state.table_buckets.get(&entry.table_bucket) {
if current.active_rename_id.is_some()
|| !matches!(current.state, TableCatalogEntryState::Active | TableCatalogEntryState::Deleted)
{
return Err(TableCatalogStoreError::Conflict("table bucket has an operation in progress".to_string()));
}
entry = current.clone();
}
let bucket = &entry.table_bucket;
if state.namespaces.keys().any(|(owner, _)| owner == bucket)
|| state.tables.keys().any(|(owner, _, _)| owner == bucket)
|| state.views.keys().any(|(owner, _, _)| owner == bucket)
|| state.commits.keys().any(|(owner, _, _)| owner == bucket)
|| state.idempotency.keys().any(|(owner, _, _)| owner == bucket)
{
return Err(TableCatalogStoreError::Conflict(
"table bucket contains namespaces or retained catalog resources".to_string(),
));
}
entry.state = TableCatalogEntryState::Deleted;
let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state);
draft_state.table_buckets.insert(entry.table_bucket.clone(), entry.clone());
(
Self::snapshot_from_mutated_state_locked(&mut draft_state, self.snapshot_write_version)?,
precondition,
StrongSnapshotWritePostcondition::TableBucketPresent(entry),
)
};
self.finalize_snapshot_write(snapshot, precondition, postcondition).await
}
async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> {
let _migration_guard = self.acquire_snapshot_write_permit().await?;
let _write_guard = self.write_lock.lock().await;
+7
View File
@@ -1343,6 +1343,13 @@ pub(crate) struct TestTableCatalogStore {
#[async_trait::async_trait]
impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore {
async fn ensure_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> {
if self.get_table_bucket(&entry.table_bucket).await?.is_some() {
return Ok(());
}
self.put_table_bucket(entry).await
}
async fn get_table_bucket(
&self,
table_bucket: &str,
+109
View File
@@ -19750,3 +19750,112 @@ async fn configured_object_catalog_dispatches_table_rename() {
Err(TableCatalogStoreError::NotFound(_))
);
}
async fn assert_disable_empty_table_bucket<S: TableCatalogStore>(store: &S) {
let bucket = "disable-empty";
let namespace = Namespace::parse("sales").unwrap();
store.ensure_table_bucket(test_bucket_entry(bucket)).await.unwrap();
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.unwrap();
assert_matches!(
store.disable_empty_table_bucket(test_bucket_entry(bucket)).await,
Err(TableCatalogStoreError::Conflict(_))
);
assert_eq!(
store.get_table_bucket(bucket).await.unwrap().unwrap().state,
TableCatalogEntryState::Active
);
store.drop_namespace(bucket, "sales").await.unwrap();
store.disable_empty_table_bucket(test_bucket_entry(bucket)).await.unwrap();
store.disable_empty_table_bucket(test_bucket_entry(bucket)).await.unwrap();
assert_eq!(
store.get_table_bucket(bucket).await.unwrap().unwrap().state,
TableCatalogEntryState::Deleted
);
// A request that observed the old enabled marker cannot reactivate the entry.
assert_matches!(
store.ensure_table_bucket(test_bucket_entry(bucket)).await,
Err(TableCatalogStoreError::NotFound(_))
);
assert_matches!(
store.create_namespace(test_namespace_entry(bucket, &namespace)).await,
Err(TableCatalogStoreError::NotFound(_))
);
store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
store
.create_namespace(test_namespace_entry(bucket, &namespace))
.await
.unwrap();
}
#[tokio::test]
async fn disable_empty_table_bucket_supports_both_catalog_backings() {
assert_disable_empty_table_bucket(&ObjectTableCatalogStore::new(TestCatalogObjectBackend::default())).await;
assert_disable_empty_table_bucket(&StrongTableCatalogStore::new(TestCatalogObjectBackend::default())).await;
}
#[tokio::test]
async fn disable_empty_table_bucket_survives_restart_and_blocks_stale_initialization() {
for mode in [TableCatalogBackingMode::ObjectBacked, TableCatalogBackingMode::DurableStrong] {
let backend = TestCatalogObjectBackend::default();
let store = ConfiguredTableCatalogStore::new_for_test(backend.clone(), mode);
// Cover an interrupted enable that wrote the bucket marker but no entry.
store.disable_empty_table_bucket(test_bucket_entry("disabled")).await.unwrap();
let restarted = ConfiguredTableCatalogStore::new_for_test(backend, mode);
assert_eq!(
restarted.get_table_bucket("disabled").await.unwrap().unwrap().state,
TableCatalogEntryState::Deleted
);
assert_matches!(
restarted.ensure_table_bucket(test_bucket_entry("disabled")).await,
Err(TableCatalogStoreError::NotFound(_))
);
}
}
#[tokio::test]
async fn disable_empty_table_bucket_serializes_with_namespace_creation() {
for mode in [TableCatalogBackingMode::ObjectBacked, TableCatalogBackingMode::DurableStrong] {
let store = ConfiguredTableCatalogStore::new_for_test(TestCatalogObjectBackend::default(), mode);
let bucket = "disable-race";
let namespace = Namespace::parse("sales").unwrap();
store.ensure_table_bucket(test_bucket_entry(bucket)).await.unwrap();
let (create, disable) = tokio::join!(
store.create_namespace(test_namespace_entry(bucket, &namespace)),
store.disable_empty_table_bucket(test_bucket_entry(bucket)),
);
assert_ne!(create.is_ok(), disable.is_ok());
if disable.is_ok() {
assert!(store.list_namespaces(bucket).await.unwrap().is_empty());
} else {
assert_eq!(
store.get_table_bucket(bucket).await.unwrap().unwrap().state,
TableCatalogEntryState::Active
);
}
}
}
#[tokio::test]
async fn disable_empty_table_bucket_failure_keeps_active_catalog() {
for mode in [TableCatalogBackingMode::ObjectBacked, TableCatalogBackingMode::DurableStrong] {
let backend = TestCatalogObjectBackend::default();
let store = ConfiguredTableCatalogStore::new_for_test(backend.clone(), mode);
let bucket = "disable-failure";
store.ensure_table_bucket(test_bucket_entry(bucket)).await.unwrap();
let path = match mode {
TableCatalogBackingMode::ObjectBacked => TableCatalogObjectPaths::default().table_bucket_entry_path(bucket),
TableCatalogBackingMode::DurableStrong => StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path(),
};
*backend.fail_put_object_path.lock().await = Some(path);
assert!(store.disable_empty_table_bucket(test_bucket_entry(bucket)).await.is_err());
let restarted = ConfiguredTableCatalogStore::new_for_test(backend, mode);
assert_eq!(
restarted.get_table_bucket(bucket).await.unwrap().unwrap().state,
TableCatalogEntryState::Active
);
restarted.disable_empty_table_bucket(test_bucket_entry(bucket)).await.unwrap();
}
}