From cf7af17b430ac0cba9834af57684ef979e60bd61 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 29 Jun 2026 07:58:14 +0800 Subject: [PATCH] feat(table-catalog): wire durable strong backing config (#3971) * feat(table-catalog): wire durable strong backing config * fix(table-catalog): guard strong backing warehouse prefixes * fix(checksums): remove unused base64 decode helper --------- Co-authored-by: Henry Guo --- crates/checksums/src/base64.rs | 20 - crates/checksums/src/lib.rs | 7 +- rustfs/src/admin/handlers/table_catalog.rs | 105 +++- rustfs/src/storage/access.rs | 3 +- rustfs/src/table_catalog.rs | 568 ++++++++++++++++++++- 5 files changed, 652 insertions(+), 51 deletions(-) diff --git a/crates/checksums/src/base64.rs b/crates/checksums/src/base64.rs index d434a4f0b..dcace1ac7 100644 --- a/crates/checksums/src/base64.rs +++ b/crates/checksums/src/base64.rs @@ -14,26 +14,6 @@ #![allow(dead_code)] use base64_simd::STANDARD; -use std::error::Error; - -#[derive(Debug)] -pub(crate) struct DecodeError(base64_simd::Error); - -impl Error for DecodeError { - fn source(&self) -> Option<&(dyn Error + 'static)> { - Some(&self.0) - } -} - -impl std::fmt::Display for DecodeError { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "failed to decode base64") - } -} - -pub(crate) fn decode(input: impl AsRef) -> Result, DecodeError> { - STANDARD.decode_to_vec(input.as_ref()).map_err(DecodeError) -} pub(crate) fn encode(input: impl AsRef<[u8]>) -> String { STANDARD.encode_to_string(input.as_ref()) diff --git a/crates/checksums/src/lib.rs b/crates/checksums/src/lib.rs index 3117a6275..c93c59798 100644 --- a/crates/checksums/src/lib.rs +++ b/crates/checksums/src/lib.rs @@ -338,8 +338,7 @@ mod tests { use crate::ChecksumAlgorithm; use crate::http::HttpChecksum; - - use crate::base64; + use base64_simd::STANDARD; use http::HeaderValue; use pretty_assertions::assert_eq; use std::fmt::Write; @@ -347,7 +346,9 @@ mod tests { const TEST_DATA: &str = r#"test data"#; fn base64_encoded_checksum_to_hex_string(header_value: &HeaderValue) -> String { - let decoded_checksum = base64::decode(header_value.to_str().unwrap()).unwrap(); + let decoded_checksum = STANDARD + .decode_to_vec(header_value.to_str().expect("checksum header value should be ASCII")) + .expect("checksum header value should be valid base64"); let decoded_checksum = decoded_checksum.into_iter().fold(String::new(), |mut acc, byte| { write!(acc, "{byte:02X?}").expect("string will always be writable"); acc diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 6efb43cab..c75f13099 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -52,6 +52,7 @@ const MAX_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60 * 60; const WAREHOUSE_PROPERTY: &str = "warehouse"; const CATALOG_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-endpoint-prefix"; const CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-compat-endpoint-prefix"; +const CATALOG_BACKING_CONFIG_KEY: &str = "rustfs.catalog-backing"; const CREDENTIAL_VENDING_CONFIG_KEY: &str = "rustfs.credential-vending"; const CREDENTIAL_VENDING_REASON_CONFIG_KEY: &str = "rustfs.credential-vending-reason"; const CREDENTIAL_SCOPE_CONFIG_KEY: &str = "rustfs.credential-scope"; @@ -988,22 +989,28 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix Ok(()) } -fn catalog_config_response() -> CatalogConfigResponse { +fn catalog_config_response() -> S3Result { let usecase = default_admin_usecase(); - CatalogConfigResponse { + let backing_mode = crate::table_catalog::TableCatalogBackingMode::from_env().map_err(catalog_store_error)?; + let mut overrides = BTreeMap::new(); + if backing_mode != crate::table_catalog::TableCatalogBackingMode::ObjectBacked { + overrides.insert(CATALOG_BACKING_CONFIG_KEY, backing_mode.as_str()); + } + Ok(CatalogConfigResponse { defaults: BTreeMap::from([ (WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID), (CATALOG_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_PREFIX), (CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_COMPAT_PREFIX), + (CATALOG_BACKING_CONFIG_KEY, crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT), ]), - overrides: BTreeMap::new(), + overrides, endpoints: TABLE_CATALOG_ENDPOINTS.to_vec(), admin_discovery: CatalogAdminDiscovery { runtime_capabilities: usecase.runtime_capabilities_route(), cluster_snapshot: usecase.cluster_snapshot_route(), extensions_catalog: usecase.extensions_catalog_route(), }, - } + }) } fn build_json_response(status: StatusCode, body: &T) -> S3Result> { @@ -1262,9 +1269,32 @@ fn table_catalog_backend() -> S3Result>; + +fn table_catalog_store_from_backend( + backend: crate::table_catalog::EcStoreTableCatalogObjectBackend, +) -> S3Result> { + crate::table_catalog::ConfiguredTableCatalogStore::from_env(backend).map_err(catalog_store_error) +} + fn table_catalog_store() -> S3Result> { let backend = table_catalog_backend()?; - Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend)) + table_catalog_store_from_backend(backend) +} + +fn table_catalog_object_store() -> S3Result { + match crate::table_catalog::TableCatalogBackingMode::from_env().map_err(catalog_store_error)? { + crate::table_catalog::TableCatalogBackingMode::ObjectBacked => { + let backend = table_catalog_backend()?; + Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend)) + } + crate::table_catalog::TableCatalogBackingMode::DurableStrong => Err(s3_error!( + InvalidRequest, + "operation is not supported with {} table catalog backing", + crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG + )), + } } async fn table_bucket_enabled_from_metadata(bucket: &str) -> S3Result { @@ -4501,8 +4531,8 @@ where }) } -async fn catalog_import_response( - store: &crate::table_catalog::ObjectTableCatalogStore, +async fn catalog_import_response( + store: &S, metadata_backend: &B, bucket: &str, namespace: &crate::table_catalog::Namespace, @@ -4511,6 +4541,7 @@ async fn catalog_import_response( table_bucket_enabled: bool, ) -> S3Result where + S: crate::table_catalog::TableCatalogStore + ?Sized, B: crate::table_catalog::TableCatalogObjectBackend, { let started = Instant::now(); @@ -4601,7 +4632,7 @@ pub struct GetCatalogConfigHandler {} impl Operation for GetCatalogConfigHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize_table_catalog_request(&req, AdminAction::GetTableCatalogAction).await?; - build_json_response(StatusCode::OK, &catalog_config_response()) + build_json_response(StatusCode::OK, &catalog_config_response()?) } } @@ -4739,7 +4770,7 @@ impl Operation for RestCreateTableHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; let response = create_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?; @@ -4758,7 +4789,7 @@ impl Operation for RestRegisterTableHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; let response = register_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?; @@ -4793,7 +4824,7 @@ impl Operation for RestCreateViewHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; let response = create_view_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?; @@ -4813,7 +4844,7 @@ impl Operation for RestLoadTableHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?; build_json_response(StatusCode::OK, &response) } @@ -4867,7 +4898,7 @@ impl Operation for RestCommitTableHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = commit_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; build_json_response(StatusCode::OK, &response) } @@ -4902,7 +4933,7 @@ impl Operation for RestLoadViewHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = load_view_response(&store, &metadata_backend, &warehouse, &namespace, &view).await?; build_json_response(StatusCode::OK, &response) } @@ -4937,7 +4968,7 @@ impl Operation for RestReplaceViewHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = replace_view_response(&store, &metadata_backend, &warehouse, &namespace, &view, request).await?; build_json_response(StatusCode::OK, &response) } @@ -4972,7 +5003,7 @@ impl Operation for ListTableRefsHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = table_refs_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?; build_json_response(StatusCode::OK, &response) } @@ -4992,7 +5023,7 @@ impl Operation for PutTableRefHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = put_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?; build_json_response(StatusCode::OK, &response) @@ -5013,7 +5044,7 @@ impl Operation for DeleteTableRefHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body_or_default::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = delete_table_ref_response(&store, &metadata_backend, &warehouse, &namespace, &table, &ref_name, request).await?; build_json_response(StatusCode::OK, &response) @@ -5050,7 +5081,7 @@ impl Operation for UpdateTableMetadataLocationHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = update_table_metadata_location_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; build_json_response(StatusCode::OK, &response) @@ -5070,7 +5101,7 @@ impl Operation for RestTableMetadataMaintenanceHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_object_store()?; let response = table_metadata_maintenance_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; build_json_response(StatusCode::OK, &response) @@ -5232,7 +5263,7 @@ impl Operation for ImportTableCatalogHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; let response = catalog_import_response(&store, &metadata_backend, &warehouse, &namespace, &table, request, table_bucket_enabled) @@ -5252,7 +5283,7 @@ impl Operation for ExternalCatalogBridgeHandler { let resource = TableCatalogResource::table(&warehouse, &namespace, &table); authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; ensure_table_bucket_enabled(&warehouse).await?; - let store = table_catalog_store()?; + let store = table_catalog_object_store()?; let response = external_catalog_bridge_response(&store, &warehouse, &namespace, &table).await?; build_json_response(StatusCode::OK, &response) } @@ -5270,7 +5301,7 @@ impl Operation for PutExternalCatalogBridgeHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; - let store = table_catalog_store()?; + let store = table_catalog_object_store()?; let response = put_external_catalog_bridge_response(&store, &warehouse, &namespace, &table, request).await?; build_json_response(StatusCode::OK, &response) } @@ -5288,7 +5319,7 @@ impl Operation for SyncExternalCatalogBridgeHandler { authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_object_store()?; if store .load_table(&warehouse, &namespace.public_name(), &table) .await @@ -5383,7 +5414,7 @@ impl Operation for RollbackTableCatalogHandler { ensure_table_bucket_enabled(&warehouse).await?; let request = read_json_body::(req.input).await?; let metadata_backend = table_catalog_backend()?; - let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let store = table_catalog_store_from_backend(metadata_backend.clone())?; let response = rollback_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; build_json_response(StatusCode::OK, &response) } @@ -5396,8 +5427,10 @@ mod tests { use std::sync::Arc; #[test] + #[serial_test::serial] fn catalog_config_response_lists_standard_rest_endpoints() { - let response = catalog_config_response(); + let response = temp_env::with_var_unset(crate::table_catalog::ENV_TABLE_CATALOG_BACKING, catalog_config_response) + .expect("catalog config should build"); assert_eq!(response.defaults.get(WAREHOUSE_PROPERTY), Some(&DEFAULT_WAREHOUSE_ID)); assert_eq!(response.defaults.get(CATALOG_ENDPOINT_PREFIX_CONFIG_KEY), Some(&TABLE_CATALOG_PREFIX)); @@ -5405,6 +5438,10 @@ mod tests { response.defaults.get(CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY), Some(&TABLE_CATALOG_COMPAT_PREFIX) ); + assert_eq!( + response.defaults.get(CATALOG_BACKING_CONFIG_KEY), + Some(&crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT) + ); assert!(response.overrides.is_empty()); assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); @@ -5508,6 +5545,22 @@ mod tests { ); } + #[test] + #[serial_test::serial] + fn catalog_config_response_reports_durable_strong_backing_override() { + let response = temp_env::with_var( + crate::table_catalog::ENV_TABLE_CATALOG_BACKING, + Some(crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG), + catalog_config_response, + ) + .expect("catalog config should build"); + + assert_eq!( + response.overrides.get(CATALOG_BACKING_CONFIG_KEY), + Some(&crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG) + ); + } + #[test] fn table_catalog_admin_operation_result_labels_are_stable() { let success: Result<(), ()> = Ok(()); diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 6b6c84908..29028ed08 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -787,7 +787,8 @@ fn table_catalog_store_for_data_plane() -> S3Result = Result; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum TableCatalogBackingMode { + ObjectBacked, + DurableStrong, +} + +impl TableCatalogBackingMode { + pub(crate) fn from_env() -> TableCatalogStoreResult { + match std::env::var(ENV_TABLE_CATALOG_BACKING) { + Ok(value) => Self::parse(&value), + Err(std::env::VarError::NotPresent) => Ok(Self::ObjectBacked), + Err(std::env::VarError::NotUnicode(_)) => Err(TableCatalogStoreError::Invalid(format!( + "{ENV_TABLE_CATALOG_BACKING} must be valid UTF-8" + ))), + } + } + + fn parse(value: &str) -> TableCatalogStoreResult { + match value.trim() { + "" | TABLE_CATALOG_BACKING_OBJECT => Ok(Self::ObjectBacked), + TABLE_CATALOG_BACKING_DURABLE_STRONG => Ok(Self::DurableStrong), + value => Err(TableCatalogStoreError::Invalid(format!( + "unsupported table catalog backing {value}; expected {TABLE_CATALOG_BACKING_OBJECT} or {TABLE_CATALOG_BACKING_DURABLE_STRONG}" + ))), + } + } + + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::ObjectBacked => TABLE_CATALOG_BACKING_OBJECT, + Self::DurableStrong => TABLE_CATALOG_BACKING_DURABLE_STRONG, + } + } +} + fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option) -> TableCatalogStoreResult { let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix); if object_prefix.is_empty() { @@ -2041,6 +2079,34 @@ where Ok(()) } + fn ensure_table_warehouse_prefix_available_locked( + state: &StrongTableCatalogState, + candidate: &TableEntry, + candidate_key: &StrongResourceKey, + ) -> TableCatalogStoreResult<()> { + if candidate.state != TableCatalogEntryState::Active { + return Ok(()); + } + let candidate_prefix = table_warehouse_object_prefix(candidate)?; + for (existing_key, existing) in &state.tables { + if existing_key == candidate_key + || existing.table_bucket != candidate.table_bucket + || existing.state != TableCatalogEntryState::Active + { + continue; + } + let Ok(existing_prefix) = table_warehouse_object_prefix(existing) else { + continue; + }; + if existing_prefix == candidate_prefix { + return Err(TableCatalogStoreError::Conflict(format!( + "table warehouse location is already registered: {candidate_prefix}" + ))); + } + } + Ok(()) + } + fn table_commit_recovery_report_for_entry_locked( state: &StrongTableCatalogState, entry: &TableEntry, @@ -2225,6 +2291,7 @@ where if let Some(warehouse_location) = next_warehouse_location { next.warehouse_location = warehouse_location; } + Self::ensure_table_warehouse_prefix_available_locked(state, &next, &key)?; next.version_token = commit_log.new_version_token.clone(); next.generation = next.generation.saturating_add(1); @@ -2264,6 +2331,44 @@ where } } +#[derive(Clone)] +pub(crate) enum ConfiguredTableCatalogStore +where + B: TableCatalogObjectBackend, +{ + ObjectBacked(ObjectTableCatalogStore), + DurableStrong(StrongTableCatalogStore), +} + +impl ConfiguredTableCatalogStore +where + B: TableCatalogObjectBackend, +{ + pub(crate) fn from_env(backend: B) -> TableCatalogStoreResult { + Ok(Self::new(backend, TableCatalogBackingMode::from_env()?)) + } + + pub(crate) fn new(backend: B, mode: TableCatalogBackingMode) -> Self { + match mode { + TableCatalogBackingMode::ObjectBacked => Self::ObjectBacked(ObjectTableCatalogStore::new(backend)), + TableCatalogBackingMode::DurableStrong => Self::DurableStrong(StrongTableCatalogStore::new(backend)), + } + } + + pub(crate) fn backing_mode(&self) -> TableCatalogBackingMode { + match self { + Self::ObjectBacked(_) => TableCatalogBackingMode::ObjectBacked, + Self::DurableStrong(_) => TableCatalogBackingMode::DurableStrong, + } + } + + fn unsupported_for_durable_strong(operation: &str) -> TableCatalogStoreError { + TableCatalogStoreError::Invalid(format!( + "{operation} is not supported with {TABLE_CATALOG_BACKING_DURABLE_STRONG} table catalog backing" + )) + } +} + #[async_trait::async_trait] impl TableCatalogStore for StrongTableCatalogStore where @@ -2403,6 +2508,7 @@ where entry.table_bucket, entry.namespace, entry.table ))); } + Self::ensure_table_warehouse_prefix_available_locked(&state, &entry, &key)?; let (precondition, rollback_state) = Self::snapshot_write_context_locked(&state); state.tables.insert(key, entry); (Self::snapshot_from_state_locked(&state), precondition, rollback_state) @@ -5454,6 +5560,324 @@ where } } +#[async_trait::async_trait] +impl TableCatalogStore for ConfiguredTableCatalogStore +where + B: TableCatalogObjectBackend, +{ + async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.get_table_bucket(table_bucket).await, + Self::DurableStrong(store) => store.get_table_bucket(table_bucket).await, + } + } + + async fn put_table_bucket(&self, entry: TableBucketEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.put_table_bucket(entry).await, + Self::DurableStrong(store) => store.put_table_bucket(entry).await, + } + } + + async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.create_namespace(entry).await, + Self::DurableStrong(store) => store.create_namespace(entry).await, + } + } + + async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_namespaces(table_bucket).await, + Self::DurableStrong(store) => store.list_namespaces(table_bucket).await, + } + } + + async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.get_namespace(table_bucket, namespace).await, + Self::DurableStrong(store) => store.get_namespace(table_bucket, namespace).await, + } + } + + async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.drop_namespace(table_bucket, namespace).await, + Self::DurableStrong(store) => store.drop_namespace(table_bucket, namespace).await, + } + } + + async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.create_table(entry).await, + Self::DurableStrong(store) => store.create_table(entry).await, + } + } + + async fn register_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.register_table(entry).await, + Self::DurableStrong(store) => store.register_table(entry).await, + } + } + + async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_tables(table_bucket, namespace).await, + Self::DurableStrong(store) => store.list_tables(table_bucket, namespace).await, + } + } + + async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.load_table(table_bucket, namespace, table).await, + Self::DurableStrong(store) => store.load_table(table_bucket, namespace, table).await, + } + } + + async fn resolve_table_data_plane_resource( + &self, + table_bucket: &str, + object: &str, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.resolve_table_data_plane_resource(table_bucket, object).await, + Self::DurableStrong(store) => store.resolve_table_data_plane_resource(table_bucket, object).await, + } + } + + async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.commit_table(request).await, + Self::DurableStrong(store) => store.commit_table(request).await, + } + } + + async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.drop_table(table_bucket, namespace, table).await, + Self::DurableStrong(store) => store.drop_table(table_bucket, namespace, table).await, + } + } + + async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.create_view(entry).await, + Self::DurableStrong(store) => store.create_view(entry).await, + } + } + + async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_views(table_bucket, namespace).await, + Self::DurableStrong(store) => store.list_views(table_bucket, namespace).await, + } + } + + async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.load_view(table_bucket, namespace, view).await, + Self::DurableStrong(store) => store.load_view(table_bucket, namespace, view).await, + } + } + + async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.replace_view(request).await, + Self::DurableStrong(store) => store.replace_view(request).await, + } + } + + async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.drop_view(table_bucket, namespace, view).await, + Self::DurableStrong(store) => store.drop_view(table_bucket, namespace, view).await, + } + } + + async fn get_commit_by_id( + &self, + table_bucket: &str, + table_id: &str, + commit_id: &str, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.get_commit_by_id(table_bucket, table_id, commit_id).await, + Self::DurableStrong(store) => store.get_commit_by_id(table_bucket, table_id, commit_id).await, + } + } + + async fn get_commit_by_idempotency_key( + &self, + table_bucket: &str, + table_id: &str, + idempotency_key: &str, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => { + store + .get_commit_by_idempotency_key(table_bucket, table_id, idempotency_key) + .await + } + Self::DurableStrong(store) => { + store + .get_commit_by_idempotency_key(table_bucket, table_id, idempotency_key) + .await + } + } + } +} + +impl ConfiguredTableCatalogStore +where + B: TableCatalogObjectBackend, +{ + pub(crate) async fn get_table_maintenance_config( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.get_table_maintenance_config(table_bucket, namespace, table).await, + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance config")), + } + } + + pub(crate) async fn put_table_maintenance_config( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + config: TableMaintenanceConfig, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .put_table_maintenance_config(table_bucket, namespace, table, config) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance config")), + } + } + + pub(crate) async fn get_table_metadata_maintenance_report( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + job_id: &str, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => { + store + .get_table_metadata_maintenance_report(table_bucket, namespace, table, job_id) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance report")), + } + } + + pub(crate) async fn run_table_metadata_maintenance_worker_once( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + worker_id: String, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .run_table_metadata_maintenance_worker_once(table_bucket, namespace, table, worker_id) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance worker")), + } + } + + pub(crate) async fn heartbeat_table_metadata_maintenance_job( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + job_id: &str, + lease_id: &str, + worker_id: &str, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .heartbeat_table_metadata_maintenance_job(table_bucket, namespace, table, job_id, lease_id, worker_id) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance heartbeat")), + } + } + + pub(crate) async fn export_table_catalog_entry( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.export_table_catalog_entry(table_bucket, namespace, table).await, + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("catalog export")), + } + } + + pub(crate) async fn diagnose_table_catalog( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + retain_recent_metadata_files: usize, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .diagnose_table_catalog(table_bucket, namespace, table, retain_recent_metadata_files) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("catalog diagnostics")), + } + } + + pub(crate) async fn recover_table_commits( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.recover_table_commits(table_bucket, namespace, table).await, + Self::DurableStrong(store) => store.plan_table_commit_recovery(table_bucket, namespace, table).await, + } + } + + pub(crate) async fn get_external_catalog_bridge( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.get_external_catalog_bridge(table_bucket, namespace, table).await, + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("external catalog bridge")), + } + } + + pub(crate) async fn put_external_catalog_bridge( + &self, + entry: ExternalCatalogBridgeEntry, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => store.put_external_catalog_bridge(entry).await, + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("external catalog bridge")), + } + } +} + pub(crate) struct EcStoreTableCatalogObjectBackend { store: Arc, } @@ -5475,7 +5899,7 @@ where } } -pub(crate) type EcStoreTableCatalogStore = ObjectTableCatalogStore>; +pub(crate) type EcStoreTableCatalogStore = ConfiguredTableCatalogStore>; #[async_trait::async_trait] impl TableCatalogObjectBackend for EcStoreTableCatalogObjectBackend @@ -8947,6 +9371,33 @@ mod tests { assert!(serde_json::from_value::(value).is_err()); } + #[test] + #[serial_test::serial] + fn table_catalog_backing_mode_defaults_to_object() { + temp_env::with_var_unset(ENV_TABLE_CATALOG_BACKING, || { + assert_eq!(TableCatalogBackingMode::from_env().unwrap(), TableCatalogBackingMode::ObjectBacked); + }); + } + + #[test] + #[serial_test::serial] + fn table_catalog_backing_mode_accepts_durable_strong_value() { + temp_env::with_var(ENV_TABLE_CATALOG_BACKING, Some(TABLE_CATALOG_BACKING_DURABLE_STRONG), || { + assert_eq!(TableCatalogBackingMode::from_env().unwrap(), TableCatalogBackingMode::DurableStrong); + }); + } + + #[test] + #[serial_test::serial] + fn table_catalog_backing_mode_rejects_unknown_value() { + temp_env::with_var(ENV_TABLE_CATALOG_BACKING, Some("memory"), || { + assert!(matches!( + TableCatalogBackingMode::from_env().unwrap_err(), + TableCatalogStoreError::Invalid(_) + )); + }); + } + struct NoopTableCatalogStore; #[async_trait::async_trait] @@ -9767,6 +10218,22 @@ mod tests { assert_eq!(object_buckets, BTreeSet::from([RUSTFS_META_BUCKET])); } + #[tokio::test] + async fn configured_table_catalog_store_uses_durable_strong_snapshot() { + let backend = TestCatalogObjectBackend::default(); + let store = ConfiguredTableCatalogStore::new(backend.clone(), TableCatalogBackingMode::DurableStrong); + let bucket = "analytics"; + + assert_eq!(store.backing_mode(), TableCatalogBackingMode::DurableStrong); + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + + let snapshot_path = StrongTableCatalogStore::::snapshot_object_path(); + assert!(backend.object_exists(RUSTFS_META_BUCKET, &snapshot_path).await.unwrap()); + + let reloaded = ConfiguredTableCatalogStore::new(backend.clone(), TableCatalogBackingMode::DurableStrong); + assert!(reloaded.get_table_bucket(bucket).await.unwrap().is_some()); + } + #[tokio::test] async fn object_table_catalog_store_persists_view_entries_and_blocks_non_empty_namespace_drop() { let backend = TestCatalogObjectBackend::default(); @@ -13607,6 +14074,105 @@ mod tests { assert!(store.load_table(bucket, "sales", "orders").await.unwrap().is_none()); } + #[tokio::test] + async fn strong_catalog_backing_rejects_duplicate_table_warehouse_location_on_register() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let orders = IdentifierSegment::parse("orders").unwrap(); + let customers = IdentifierSegment::parse("customers").unwrap(); + let orders_metadata = default_table_metadata_file_path(&namespace, &orders, "00001.metadata.json"); + let customers_metadata = default_table_metadata_file_path(&namespace, &customers, "00001.metadata.json"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &orders, orders_metadata)) + .await + .unwrap(); + let mut duplicate = test_table_entry(bucket, &namespace, &customers, customers_metadata); + duplicate.table_id = "table-id-2".to_string(); + duplicate.table_uuid = "table-uuid-2".to_string(); + duplicate.warehouse_location = format!("s3://{bucket}/tables/table-id"); + + let err = store.register_table(duplicate).await.unwrap_err(); + + assert_matches!(err, TableCatalogStoreError::Conflict(_)); + assert!(store.load_table(bucket, "sales", "customers").await.unwrap().is_none()); + } + + #[tokio::test] + async fn strong_catalog_backing_rejects_duplicate_table_warehouse_location_on_commit_relocation() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let orders = IdentifierSegment::parse("orders").unwrap(); + let customers = IdentifierSegment::parse("customers").unwrap(); + let orders_metadata = default_table_metadata_file_path(&namespace, &orders, "00001.metadata.json"); + let relocated_metadata = default_table_metadata_file_path(&namespace, &orders, "00002.metadata.json"); + let customers_metadata = default_table_metadata_file_path(&namespace, &customers, "00001.metadata.json"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + store + .create_table(test_table_entry(bucket, &namespace, &orders, orders_metadata.clone())) + .await + .unwrap(); + let mut customers_entry = test_table_entry(bucket, &namespace, &customers, customers_metadata); + customers_entry.table_id = "table-id-2".to_string(); + customers_entry.table_uuid = "table-uuid-2".to_string(); + customers_entry.warehouse_location = format!("s3://{bucket}/tables/customer-id"); + store.create_table(customers_entry).await.unwrap(); + backend + .seed_object( + bucket, + &relocated_metadata, + serde_json::to_vec(&serde_json::json!({ + "location": "s3://analytics/tables/customer-id", + "table-uuid": "table-uuid" + })) + .unwrap(), + ) + .await; + + let err = store + .commit_table(TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: orders.as_str().to_string(), + commit_id: "commit-1".to_string(), + idempotency_key: Some("client-request".to_string()), + operation: "set-location".to_string(), + expected_version_token: "token-v1".to_string(), + expected_metadata_location: orders_metadata.clone(), + new_metadata_location: relocated_metadata, + requirements: Vec::new(), + writer: Some("pyiceberg/test".to_string()), + }) + .await + .unwrap_err(); + + assert_matches!(err, TableCatalogStoreError::Conflict(_)); + let loaded = store.load_table(bucket, "sales", "orders").await.unwrap().unwrap(); + assert_eq!(loaded.metadata_location, orders_metadata); + assert_eq!(loaded.warehouse_location, format!("s3://{bucket}/tables/table-id")); + assert!( + store + .get_commit_by_id(bucket, "table-id", "commit-1") + .await + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn diagnostics_backing_manifest_requires_recovery_before_migration() { let backend = TestCatalogObjectBackend::default();