mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-02 10:18:10 +00:00
feat(table-catalog): update object namespace properties (#6815)
* feat(table-catalog): update object namespace properties * test(table-catalog): cover object namespace properties * fix(ci): use admin storage contract in catalog test * fix(table-catalog): reject corrupt namespace properties
This commit is contained in:
@@ -160,6 +160,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
|
|||||||
"GET /v1/{prefix}/namespaces/{namespace}",
|
"GET /v1/{prefix}/namespaces/{namespace}",
|
||||||
"HEAD /v1/{prefix}/namespaces/{namespace}",
|
"HEAD /v1/{prefix}/namespaces/{namespace}",
|
||||||
"DELETE /v1/{prefix}/namespaces/{namespace}",
|
"DELETE /v1/{prefix}/namespaces/{namespace}",
|
||||||
|
"POST /v1/{prefix}/namespaces/{namespace}/properties",
|
||||||
"GET /v1/{prefix}/namespaces/{namespace}/tables",
|
"GET /v1/{prefix}/namespaces/{namespace}/tables",
|
||||||
"POST /v1/{prefix}/namespaces/{namespace}/tables",
|
"POST /v1/{prefix}/namespaces/{namespace}/tables",
|
||||||
"POST /v1/{prefix}/namespaces/{namespace}/register",
|
"POST /v1/{prefix}/namespaces/{namespace}/register",
|
||||||
@@ -176,7 +177,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
|
|||||||
"POST /v1/{prefix}/namespaces/{namespace}/views/{view}",
|
"POST /v1/{prefix}/namespaces/{namespace}/views/{view}",
|
||||||
"DELETE /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"];
|
const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &[];
|
||||||
|
|
||||||
static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
|
static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
|
||||||
static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {};
|
static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {};
|
||||||
|
|||||||
@@ -434,7 +434,7 @@ fn catalog_config_response_lists_standard_rest_endpoints() {
|
|||||||
Some(REST_NAMESPACE_SEPARATOR_URL_ENCODED)
|
Some(REST_NAMESPACE_SEPARATOR_URL_ENCODED)
|
||||||
);
|
);
|
||||||
assert!(
|
assert!(
|
||||||
!response
|
response
|
||||||
.endpoints
|
.endpoints
|
||||||
.contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties")
|
.contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties")
|
||||||
);
|
);
|
||||||
@@ -466,6 +466,7 @@ fn catalog_config_response_reports_durable_strong_backing_override() {
|
|||||||
.contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties")
|
.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.endpoints.as_slice(), TABLE_CATALOG_ENDPOINTS);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
@@ -11273,6 +11274,183 @@ async fn namespace_helpers_call_catalog_store() {
|
|||||||
assert!(list.namespaces.is_empty());
|
assert!(list.namespaces.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
#[serial_test::serial]
|
||||||
|
async fn namespace_property_handler_updates_object_backed_catalog_and_maps_errors() {
|
||||||
|
use crate::admin::storage_api::contract::bucket::{BucketOperations as _, MakeBucketOptions};
|
||||||
|
|
||||||
|
temp_env::async_with_vars(
|
||||||
|
[(
|
||||||
|
crate::table_catalog::ENV_TABLE_CATALOG_BACKING,
|
||||||
|
Some(crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT),
|
||||||
|
)],
|
||||||
|
async {
|
||||||
|
let (_temp_dir, _disk_paths, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
|
||||||
|
let bucket = format!("namespace-properties-{}", Uuid::new_v4().simple());
|
||||||
|
object_store
|
||||||
|
.make_bucket(&bucket, &MakeBucketOptions::default())
|
||||||
|
.await
|
||||||
|
.expect("table bucket should be created");
|
||||||
|
enable_table_bucket_marker(&object_store, &bucket)
|
||||||
|
.await
|
||||||
|
.expect("table bucket marker should be enabled");
|
||||||
|
|
||||||
|
rustfs_iam::store::object::ObjectStore::new(object_store.clone())
|
||||||
|
.save_iam_config(
|
||||||
|
serde_json::json!({"version": 1}),
|
||||||
|
format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("request IAM format should be seeded");
|
||||||
|
let iam = rustfs_iam::build_iam_sys(object_store.clone())
|
||||||
|
.await
|
||||||
|
.expect("request IAM should initialize");
|
||||||
|
let context = Arc::new(AppContext::new(
|
||||||
|
object_store.clone(),
|
||||||
|
Arc::new(RequestIam { handle: iam }),
|
||||||
|
Arc::new(RequestKms),
|
||||||
|
));
|
||||||
|
let root_access_key = "namespace-properties-root";
|
||||||
|
let root_secret_key = "namespace-properties-root-secret";
|
||||||
|
assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
|
||||||
|
access_key: root_access_key.to_string(),
|
||||||
|
secret_key: root_secret_key.to_string(),
|
||||||
|
status: "on".to_string(),
|
||||||
|
..Default::default()
|
||||||
|
}));
|
||||||
|
let slot = ServerContextSlot::new();
|
||||||
|
assert!(slot.install(context.clone()));
|
||||||
|
|
||||||
|
let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
|
||||||
|
object_store,
|
||||||
|
context.table_catalog_strong_runtime(),
|
||||||
|
);
|
||||||
|
let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
|
||||||
|
backend.clone(),
|
||||||
|
crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
|
||||||
|
);
|
||||||
|
catalog
|
||||||
|
.put_table_bucket(table_bucket_entry_from_metadata_marker(&bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket catalog entry should be seeded");
|
||||||
|
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||||
|
let entry = crate::table_catalog::NamespaceEntry {
|
||||||
|
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
|
||||||
|
table_bucket: bucket.clone(),
|
||||||
|
namespace: namespace.public_name(),
|
||||||
|
namespace_id: namespace.storage_id(),
|
||||||
|
state: crate::table_catalog::TableCatalogEntryState::Active,
|
||||||
|
properties: BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]),
|
||||||
|
created_at: None,
|
||||||
|
updated_at: None,
|
||||||
|
};
|
||||||
|
catalog
|
||||||
|
.create_namespace(entry.clone())
|
||||||
|
.await
|
||||||
|
.expect("namespace should be seeded");
|
||||||
|
|
||||||
|
let request = |namespace: &str, body: serde_json::Value| {
|
||||||
|
let mut extensions = http::Extensions::new();
|
||||||
|
extensions.insert(slot.clone());
|
||||||
|
S3Request {
|
||||||
|
input: Body::from(serde_json::to_vec(&body).expect("request body should serialize")),
|
||||||
|
method: Method::POST,
|
||||||
|
uri: format!("/iceberg/v1/{bucket}/namespaces/{namespace}/properties")
|
||||||
|
.parse()
|
||||||
|
.expect("request URI should parse"),
|
||||||
|
headers: HeaderMap::new(),
|
||||||
|
extensions,
|
||||||
|
credentials: Some(s3s::auth::Credentials {
|
||||||
|
access_key: root_access_key.to_string(),
|
||||||
|
secret_key: s3s::auth::SecretKey::from(root_secret_key.to_string()),
|
||||||
|
}),
|
||||||
|
region: None,
|
||||||
|
service: None,
|
||||||
|
trailing_headers: None,
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let mut params_router = matchit::Router::new();
|
||||||
|
params_router
|
||||||
|
.insert("/iceberg/v1/{warehouse}/namespaces/{namespace}/properties", ())
|
||||||
|
.expect("handler parameter route should register");
|
||||||
|
let success_path = format!("/iceberg/v1/{bucket}/namespaces/analytics/properties");
|
||||||
|
let params = params_router
|
||||||
|
.at(&success_path)
|
||||||
|
.expect("success handler parameters should match")
|
||||||
|
.params;
|
||||||
|
let response = RestUpdateNamespacePropertiesHandler {}
|
||||||
|
.call(
|
||||||
|
request(
|
||||||
|
"analytics",
|
||||||
|
serde_json::json!({
|
||||||
|
"removals": ["owner", "missing"],
|
||||||
|
"updates": {"retention": "30d"}
|
||||||
|
}),
|
||||||
|
),
|
||||||
|
params,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("handler should update object-backed namespace properties");
|
||||||
|
assert_eq!(response.output.0, StatusCode::OK);
|
||||||
|
let body = http_body_util::BodyExt::collect(response.output.1)
|
||||||
|
.await
|
||||||
|
.expect("response body should collect")
|
||||||
|
.to_bytes();
|
||||||
|
assert_eq!(
|
||||||
|
serde_json::from_slice::<serde_json::Value>(&body).expect("response body should decode"),
|
||||||
|
serde_json::json!({
|
||||||
|
"updated": ["retention"],
|
||||||
|
"removed": ["owner"],
|
||||||
|
"missing": ["missing"]
|
||||||
|
})
|
||||||
|
);
|
||||||
|
let persisted = catalog
|
||||||
|
.get_namespace(&bucket, &namespace.public_name())
|
||||||
|
.await
|
||||||
|
.expect("updated namespace should load")
|
||||||
|
.expect("updated namespace should remain");
|
||||||
|
assert_eq!(persisted.properties.get("retention").map(String::as_str), Some("30d"));
|
||||||
|
assert!(!persisted.properties.contains_key("owner"));
|
||||||
|
|
||||||
|
let missing_path = format!("/iceberg/v1/{bucket}/namespaces/missing/properties");
|
||||||
|
let params = params_router
|
||||||
|
.at(&missing_path)
|
||||||
|
.expect("missing handler parameters should match")
|
||||||
|
.params;
|
||||||
|
let missing = RestUpdateNamespacePropertiesHandler {}
|
||||||
|
.call(request("missing", serde_json::json!({"updates": {"owner": "platform"}})), params)
|
||||||
|
.await
|
||||||
|
.expect_err("missing namespace should fail");
|
||||||
|
assert_eq!(missing.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_NO_SUCH_NAMESPACE.into()));
|
||||||
|
assert_eq!(missing.status_code(), Some(StatusCode::NOT_FOUND));
|
||||||
|
|
||||||
|
let corrupt = crate::table_catalog::Namespace::parse("corrupt").expect("namespace should parse");
|
||||||
|
let corrupt_path = crate::table_catalog::TableCatalogObjectPaths::default().namespace_entry_path(&bucket, &corrupt);
|
||||||
|
backend
|
||||||
|
.put_object(
|
||||||
|
crate::admin::storage_api::RUSTFS_META_BUCKET,
|
||||||
|
&corrupt_path,
|
||||||
|
b"{".to_vec(),
|
||||||
|
crate::table_catalog::TableCatalogPutPrecondition::Any,
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("corrupt namespace entry should be seeded");
|
||||||
|
let corrupt_request_path = format!("/iceberg/v1/{bucket}/namespaces/corrupt/properties");
|
||||||
|
let params = params_router
|
||||||
|
.at(&corrupt_request_path)
|
||||||
|
.expect("corrupt handler parameters should match")
|
||||||
|
.params;
|
||||||
|
let corrupt = RestUpdateNamespacePropertiesHandler {}
|
||||||
|
.call(request("corrupt", serde_json::json!({"updates": {"owner": "platform"}})), params)
|
||||||
|
.await
|
||||||
|
.expect_err("corrupt namespace should fail");
|
||||||
|
assert_eq!(corrupt.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into()));
|
||||||
|
assert_eq!(corrupt.status_code(), Some(StatusCode::BAD_REQUEST));
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn table_helpers_call_catalog_store() {
|
async fn table_helpers_call_catalog_store() {
|
||||||
let store = TestTableCatalogStore::default();
|
let store = TestTableCatalogStore::default();
|
||||||
|
|||||||
@@ -1171,9 +1171,7 @@ where
|
|||||||
update: NamespacePropertiesUpdate,
|
update: NamespacePropertiesUpdate,
|
||||||
) -> TableCatalogStoreResult<NamespacePropertiesUpdateResult> {
|
) -> TableCatalogStoreResult<NamespacePropertiesUpdateResult> {
|
||||||
match self {
|
match self {
|
||||||
Self::ObjectBacked(_) => Err(TableCatalogStoreError::Unsupported(
|
Self::ObjectBacked(store) => store.update_namespace_properties(table_bucket, namespace, update).await,
|
||||||
"namespace property updates require durable-strong catalog backing".to_string(),
|
|
||||||
)),
|
|
||||||
Self::DurableStrong(store) => store.update_namespace_properties(table_bucket, namespace, update).await,
|
Self::DurableStrong(store) => store.update_namespace_properties(table_bucket, namespace, update).await,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4636,6 +4636,66 @@ where
|
|||||||
Ok(None)
|
Ok(None)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn update_namespace_properties(
|
||||||
|
&self,
|
||||||
|
table_bucket: &str,
|
||||||
|
namespace: &str,
|
||||||
|
update: NamespacePropertiesUpdate,
|
||||||
|
) -> TableCatalogStoreResult<NamespacePropertiesUpdateResult> {
|
||||||
|
let namespace = parse_namespace_for_store(namespace)?;
|
||||||
|
self.require_table_bucket(table_bucket).await?;
|
||||||
|
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?;
|
||||||
|
let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace);
|
||||||
|
let _namespace_guard = self
|
||||||
|
.backend
|
||||||
|
.acquire_write_lock(self.catalog_bucket(), &namespace_path)
|
||||||
|
.await?;
|
||||||
|
|
||||||
|
let current = self
|
||||||
|
.read_entry_unlocked::<NamespaceEntry>(self.catalog_bucket(), &namespace_path)
|
||||||
|
.await?;
|
||||||
|
let (mut next, precondition) = match current {
|
||||||
|
Some((entry, etag)) => {
|
||||||
|
validate_namespace_entry_object(&self.paths, &namespace_path, &entry)?;
|
||||||
|
validate_namespace_properties(&entry.properties)?;
|
||||||
|
if entry.state != TableCatalogEntryState::Active {
|
||||||
|
return Err(TableCatalogStoreError::NotFound(format!(
|
||||||
|
"namespace {table_bucket}/{}",
|
||||||
|
namespace.public_name()
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
let etag = etag.ok_or_else(|| {
|
||||||
|
TableCatalogStoreError::Internal(format!("catalog namespace entry has no etag: {namespace_path}"))
|
||||||
|
})?;
|
||||||
|
(entry, TableCatalogPutPrecondition::IfMatch(etag))
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
if !self.has_active_namespace_object(table_bucket, &namespace).await?
|
||||||
|
&& !self.has_active_namespace_descendant(table_bucket, &namespace).await?
|
||||||
|
{
|
||||||
|
return Err(TableCatalogStoreError::NotFound(format!(
|
||||||
|
"namespace {table_bucket}/{}",
|
||||||
|
namespace.public_name()
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
(synthetic_namespace_entry(table_bucket, &namespace), TableCatalogPutPrecondition::IfAbsent)
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let before = next.clone();
|
||||||
|
let result = update.apply_to(&mut next);
|
||||||
|
validate_namespace_entry_object(&self.paths, &namespace_path, &next)?;
|
||||||
|
validate_namespace_properties(&next.properties)?;
|
||||||
|
if before == next {
|
||||||
|
return Ok(result);
|
||||||
|
}
|
||||||
|
self.write_entry_unlocked(self.catalog_bucket(), &namespace_path, &next, precondition)
|
||||||
|
.await?;
|
||||||
|
Ok(result)
|
||||||
|
}
|
||||||
|
|
||||||
async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
|
async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
|
||||||
let parent = parse_namespace_for_store(parent)?;
|
let parent = parse_namespace_for_store(parent)?;
|
||||||
let prefix = format!("{}{}/", self.paths.namespace_entries_prefix(table_bucket), parent.storage_id());
|
let prefix = format!("{}{}/", self.paths.namespace_entries_prefix(table_bucket), parent.storage_id());
|
||||||
|
|||||||
@@ -16617,13 +16617,16 @@ fn namespace_property_update_and_limits_reject_ambiguous_or_oversized_state() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn configured_object_catalog_rejects_namespace_property_update_without_mutation() {
|
async fn configured_object_catalog_updates_namespace_properties() {
|
||||||
let backend = TestCatalogObjectBackend::default();
|
let backend = TestCatalogObjectBackend::default();
|
||||||
let store = ConfiguredTableCatalogStore::new_for_test(backend, TableCatalogBackingMode::ObjectBacked);
|
let store = ConfiguredTableCatalogStore::new_for_test(backend, TableCatalogBackingMode::ObjectBacked);
|
||||||
let bucket = "analytics";
|
let bucket = "analytics";
|
||||||
let namespace = Namespace::parse("sales").expect("namespace should parse");
|
let namespace = Namespace::parse("sales").expect("namespace should parse");
|
||||||
let mut entry = test_namespace_entry(bucket, &namespace);
|
let mut entry = test_namespace_entry(bucket, &namespace);
|
||||||
entry.properties = BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]);
|
entry.properties = BTreeMap::from([
|
||||||
|
("obsolete".to_string(), "true".to_string()),
|
||||||
|
("owner".to_string(), "lakehouse".to_string()),
|
||||||
|
]);
|
||||||
store
|
store
|
||||||
.put_table_bucket(test_bucket_entry(bucket))
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
.await
|
.await
|
||||||
@@ -16634,18 +16637,354 @@ async fn configured_object_catalog_rejects_namespace_property_update_without_mut
|
|||||||
.update_namespace_properties(
|
.update_namespace_properties(
|
||||||
bucket,
|
bucket,
|
||||||
"sales",
|
"sales",
|
||||||
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]))
|
NamespacePropertiesUpdate::try_new(
|
||||||
.expect("namespace update should validate"),
|
vec!["obsolete".to_string(), "missing".to_string()],
|
||||||
|
BTreeMap::from([
|
||||||
|
("owner".to_string(), "platform".to_string()),
|
||||||
|
("retention".to_string(), "30d".to_string()),
|
||||||
|
]),
|
||||||
|
)
|
||||||
|
.expect("namespace update should validate"),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.expect_err("object-backed namespace property update should be unsupported");
|
.expect("object-backed namespace properties should update");
|
||||||
assert_matches!(result, TableCatalogStoreError::Unsupported(_));
|
assert_eq!(result.updated, vec!["owner".to_string(), "retention".to_string()]);
|
||||||
|
assert_eq!(result.removed, vec!["obsolete".to_string()]);
|
||||||
|
assert_eq!(result.missing, vec!["missing".to_string()]);
|
||||||
let stored = store
|
let stored = store
|
||||||
.get_namespace(bucket, "sales")
|
.get_namespace(bucket, "sales")
|
||||||
.await
|
.await
|
||||||
.expect("namespace lookup should succeed")
|
.expect("namespace lookup should succeed")
|
||||||
.expect("namespace should remain");
|
.expect("namespace should remain");
|
||||||
assert_eq!(stored.properties.get("owner").map(String::as_str), Some("lakehouse"));
|
assert_eq!(
|
||||||
|
stored.properties,
|
||||||
|
BTreeMap::from([
|
||||||
|
("owner".to_string(), "platform".to_string()),
|
||||||
|
("retention".to_string(), "30d".to_string()),
|
||||||
|
])
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn object_catalog_namespace_property_update_materializes_implicit_parent() {
|
||||||
|
let backend = TestCatalogObjectBackend::default();
|
||||||
|
let store = ObjectTableCatalogStore::new(backend);
|
||||||
|
let bucket = "analytics";
|
||||||
|
let parent = Namespace::parse("sales").expect("parent namespace should parse");
|
||||||
|
let child = Namespace::parse("sales.daily").expect("child namespace should parse");
|
||||||
|
store
|
||||||
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket entry should be seeded");
|
||||||
|
let child_entry = test_namespace_entry(bucket, &child);
|
||||||
|
store
|
||||||
|
.create_namespace(child_entry.clone())
|
||||||
|
.await
|
||||||
|
.expect("child namespace should be created");
|
||||||
|
|
||||||
|
let no_change = store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&parent.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(vec!["missing".to_string()], BTreeMap::new())
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("implicit parent no-op should succeed");
|
||||||
|
assert_eq!(no_change.missing, vec!["missing".to_string()]);
|
||||||
|
let parent_path = store.paths.namespace_entry_path(bucket, &parent);
|
||||||
|
assert!(
|
||||||
|
store
|
||||||
|
.read_entry::<NamespaceEntry>(store.catalog_bucket(), &parent_path)
|
||||||
|
.await
|
||||||
|
.expect("implicit parent lookup should succeed")
|
||||||
|
.is_none()
|
||||||
|
);
|
||||||
|
|
||||||
|
let result = store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&parent.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(
|
||||||
|
vec!["missing".to_string()],
|
||||||
|
BTreeMap::from([("owner".to_string(), "platform".to_string())]),
|
||||||
|
)
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("implicit parent should materialize");
|
||||||
|
|
||||||
|
assert_eq!(result.updated, vec!["owner".to_string()]);
|
||||||
|
assert!(result.removed.is_empty());
|
||||||
|
assert_eq!(result.missing, vec!["missing".to_string()]);
|
||||||
|
let (materialized, _) = store
|
||||||
|
.read_entry::<NamespaceEntry>(store.catalog_bucket(), &parent_path)
|
||||||
|
.await
|
||||||
|
.expect("materialized parent should load")
|
||||||
|
.expect("parent should have an explicit entry");
|
||||||
|
assert_eq!(materialized.properties.get("owner").map(String::as_str), Some("platform"));
|
||||||
|
assert_eq!(
|
||||||
|
store
|
||||||
|
.get_namespace(bucket, &child.public_name())
|
||||||
|
.await
|
||||||
|
.expect("child lookup should succeed"),
|
||||||
|
Some(child_entry)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn object_catalog_namespace_property_update_materializes_resource_only_parents() {
|
||||||
|
let backend = TestCatalogObjectBackend::default();
|
||||||
|
let store = ObjectTableCatalogStore::new(backend.clone());
|
||||||
|
let bucket = "analytics";
|
||||||
|
let table_namespace = Namespace::parse("table_only").expect("table namespace should parse");
|
||||||
|
let view_namespace = Namespace::parse("view_only").expect("view namespace should parse");
|
||||||
|
let table = IdentifierSegment::parse("orders").expect("table should parse");
|
||||||
|
let view = IdentifierSegment::parse("recent_orders").expect("view should parse");
|
||||||
|
store
|
||||||
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket entry should be seeded");
|
||||||
|
|
||||||
|
let table_entry = test_table_entry(
|
||||||
|
bucket,
|
||||||
|
&table_namespace,
|
||||||
|
&table,
|
||||||
|
default_table_metadata_file_path(&table_namespace, &table, "00001.metadata.json"),
|
||||||
|
);
|
||||||
|
backend
|
||||||
|
.seed_object(
|
||||||
|
RUSTFS_META_BUCKET,
|
||||||
|
&store.paths.table_entry_path(bucket, &table_namespace, &table),
|
||||||
|
serde_json::to_vec(&table_entry).expect("table entry should serialize"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let view_entry = test_view_entry(
|
||||||
|
bucket,
|
||||||
|
&view_namespace,
|
||||||
|
&view,
|
||||||
|
default_view_metadata_file_path(&view_namespace, &view, "00001.view.json"),
|
||||||
|
);
|
||||||
|
backend
|
||||||
|
.seed_object(
|
||||||
|
RUSTFS_META_BUCKET,
|
||||||
|
&store.paths.view_entry_path(bucket, &view_namespace, &view),
|
||||||
|
serde_json::to_vec(&view_entry).expect("view entry should serialize"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let table_before = store
|
||||||
|
.load_table(bucket, &table_namespace.public_name(), table.as_str())
|
||||||
|
.await
|
||||||
|
.expect("table should load before materializing its namespace");
|
||||||
|
let view_before = store
|
||||||
|
.load_view(bucket, &view_namespace.public_name(), view.as_str())
|
||||||
|
.await
|
||||||
|
.expect("view should load before materializing its namespace");
|
||||||
|
|
||||||
|
for namespace in [&table_namespace, &view_namespace] {
|
||||||
|
let result = store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&namespace.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]))
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("active resource should prove the implicit namespace");
|
||||||
|
assert_eq!(result.updated, vec!["owner".to_string()]);
|
||||||
|
let materialized = store
|
||||||
|
.get_namespace(bucket, &namespace.public_name())
|
||||||
|
.await
|
||||||
|
.expect("materialized namespace should load")
|
||||||
|
.expect("materialized namespace should exist");
|
||||||
|
assert_eq!(materialized.properties.get("owner").map(String::as_str), Some("platform"));
|
||||||
|
}
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
store
|
||||||
|
.load_table(bucket, &table_namespace.public_name(), table.as_str())
|
||||||
|
.await
|
||||||
|
.expect("table should load after materializing its namespace"),
|
||||||
|
table_before
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
store
|
||||||
|
.load_view(bucket, &view_namespace.public_name(), view.as_str())
|
||||||
|
.await
|
||||||
|
.expect("view should load after materializing its namespace"),
|
||||||
|
view_before
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn object_catalog_namespace_property_update_requires_etag_and_skips_noop_writes() {
|
||||||
|
let backend = TestCatalogObjectBackend::default();
|
||||||
|
let store = ObjectTableCatalogStore::new(backend.clone());
|
||||||
|
let bucket = "analytics";
|
||||||
|
store
|
||||||
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket entry should be seeded");
|
||||||
|
|
||||||
|
let etagless = Namespace::parse("etagless").expect("namespace should parse");
|
||||||
|
let mut etagless_entry = test_namespace_entry(bucket, &etagless);
|
||||||
|
etagless_entry.properties.insert("owner".to_string(), "lakehouse".to_string());
|
||||||
|
store
|
||||||
|
.create_namespace(etagless_entry)
|
||||||
|
.await
|
||||||
|
.expect("etagless namespace should be seeded");
|
||||||
|
let etagless_path = store.paths.namespace_entry_path(bucket, &etagless);
|
||||||
|
backend.omit_etag_for_object(RUSTFS_META_BUCKET, &etagless_path).await;
|
||||||
|
let etagless_puts = backend.put_attempt_count(RUSTFS_META_BUCKET, &etagless_path).await;
|
||||||
|
|
||||||
|
assert_matches!(
|
||||||
|
store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&etagless.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]),)
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(TableCatalogStoreError::Internal(_))
|
||||||
|
);
|
||||||
|
assert_eq!(backend.put_attempt_count(RUSTFS_META_BUCKET, &etagless_path).await, etagless_puts);
|
||||||
|
let unchanged = store
|
||||||
|
.get_namespace(bucket, &etagless.public_name())
|
||||||
|
.await
|
||||||
|
.expect("etagless namespace should still load")
|
||||||
|
.expect("etagless namespace should remain");
|
||||||
|
assert_eq!(unchanged.properties.get("owner").map(String::as_str), Some("lakehouse"));
|
||||||
|
|
||||||
|
let no_op = Namespace::parse("no_op").expect("namespace should parse");
|
||||||
|
let mut no_op_entry = test_namespace_entry(bucket, &no_op);
|
||||||
|
no_op_entry.properties.insert("owner".to_string(), "lakehouse".to_string());
|
||||||
|
store
|
||||||
|
.create_namespace(no_op_entry)
|
||||||
|
.await
|
||||||
|
.expect("no-op namespace should be seeded");
|
||||||
|
let no_op_path = store.paths.namespace_entry_path(bucket, &no_op);
|
||||||
|
backend.fail_next_put(RUSTFS_META_BUCKET, &no_op_path).await;
|
||||||
|
let puts_before_no_op = backend.put_attempt_count(RUSTFS_META_BUCKET, &no_op_path).await;
|
||||||
|
|
||||||
|
let result = store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&no_op.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]))
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("unchanged namespace properties should not write");
|
||||||
|
assert_eq!(result.updated, vec!["owner".to_string()]);
|
||||||
|
assert_eq!(backend.put_attempt_count(RUSTFS_META_BUCKET, &no_op_path).await, puts_before_no_op);
|
||||||
|
|
||||||
|
assert_matches!(
|
||||||
|
store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
&no_op.public_name(),
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]),)
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await,
|
||||||
|
Err(TableCatalogStoreError::Internal(_))
|
||||||
|
);
|
||||||
|
let unchanged = store
|
||||||
|
.get_namespace(bucket, &no_op.public_name())
|
||||||
|
.await
|
||||||
|
.expect("namespace should load after failed write")
|
||||||
|
.expect("namespace should remain");
|
||||||
|
assert_eq!(unchanged.properties.get("owner").map(String::as_str), Some("lakehouse"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn object_catalog_namespace_property_update_rejects_missing_inactive_and_corrupt_entries() {
|
||||||
|
let backend = TestCatalogObjectBackend::default();
|
||||||
|
let store = ObjectTableCatalogStore::new(backend.clone());
|
||||||
|
let bucket = "analytics";
|
||||||
|
store
|
||||||
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket entry should be seeded");
|
||||||
|
let update = || {
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())]))
|
||||||
|
.expect("namespace update should validate")
|
||||||
|
};
|
||||||
|
|
||||||
|
assert_matches!(
|
||||||
|
store.update_namespace_properties(bucket, "missing", update()).await,
|
||||||
|
Err(TableCatalogStoreError::NotFound(_))
|
||||||
|
);
|
||||||
|
|
||||||
|
let inactive = Namespace::parse("inactive").expect("inactive namespace should parse");
|
||||||
|
let mut inactive_entry = test_namespace_entry(bucket, &inactive);
|
||||||
|
inactive_entry.state = TableCatalogEntryState::Deleted;
|
||||||
|
backend
|
||||||
|
.seed_object(
|
||||||
|
RUSTFS_META_BUCKET,
|
||||||
|
&store.paths.namespace_entry_path(bucket, &inactive),
|
||||||
|
serde_json::to_vec(&inactive_entry).expect("inactive namespace should encode"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
assert_matches!(
|
||||||
|
store
|
||||||
|
.update_namespace_properties(bucket, &inactive.public_name(), update())
|
||||||
|
.await,
|
||||||
|
Err(TableCatalogStoreError::NotFound(_))
|
||||||
|
);
|
||||||
|
|
||||||
|
let corrupt = Namespace::parse("corrupt").expect("corrupt namespace should parse");
|
||||||
|
backend
|
||||||
|
.seed_object(RUSTFS_META_BUCKET, &store.paths.namespace_entry_path(bucket, &corrupt), b"{".to_vec())
|
||||||
|
.await;
|
||||||
|
assert_matches!(
|
||||||
|
store
|
||||||
|
.update_namespace_properties(bucket, &corrupt.public_name(), update())
|
||||||
|
.await,
|
||||||
|
Err(TableCatalogStoreError::Invalid(_))
|
||||||
|
);
|
||||||
|
|
||||||
|
let semantically_corrupt = Namespace::parse("semantically_corrupt").expect("corrupt namespace should parse");
|
||||||
|
let mut semantically_corrupt_entry = test_namespace_entry(bucket, &semantically_corrupt);
|
||||||
|
semantically_corrupt_entry.properties = (0..=NAMESPACE_PROPERTIES_MAX_ENTRIES)
|
||||||
|
.map(|index| (format!("key{index}"), "value".to_string()))
|
||||||
|
.collect();
|
||||||
|
let semantically_corrupt_path = store.paths.namespace_entry_path(bucket, &semantically_corrupt);
|
||||||
|
backend
|
||||||
|
.seed_object(
|
||||||
|
RUSTFS_META_BUCKET,
|
||||||
|
&semantically_corrupt_path,
|
||||||
|
serde_json::to_vec(&semantically_corrupt_entry).expect("corrupt namespace should encode"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let put_attempts = backend
|
||||||
|
.put_attempt_count(RUSTFS_META_BUCKET, &semantically_corrupt_path)
|
||||||
|
.await;
|
||||||
|
let repair_update =
|
||||||
|
NamespacePropertiesUpdate::try_new(vec![format!("key{NAMESPACE_PROPERTIES_MAX_ENTRIES}")], BTreeMap::new())
|
||||||
|
.expect("repair request should validate structurally");
|
||||||
|
|
||||||
|
assert_matches!(
|
||||||
|
store
|
||||||
|
.update_namespace_properties(bucket, &semantically_corrupt.public_name(), repair_update,)
|
||||||
|
.await,
|
||||||
|
Err(TableCatalogStoreError::Invalid(_))
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
backend
|
||||||
|
.put_attempt_count(RUSTFS_META_BUCKET, &semantically_corrupt_path)
|
||||||
|
.await,
|
||||||
|
put_attempts
|
||||||
|
);
|
||||||
|
let persisted = store
|
||||||
|
.read_entry::<NamespaceEntry>(RUSTFS_META_BUCKET, &semantically_corrupt_path)
|
||||||
|
.await
|
||||||
|
.expect("corrupt namespace lookup should succeed")
|
||||||
|
.expect("corrupt namespace should remain")
|
||||||
|
.0;
|
||||||
|
assert_eq!(persisted.properties.len(), NAMESPACE_PROPERTIES_MAX_ENTRIES + 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
@@ -16984,6 +17323,59 @@ async fn object_catalog_namespace_replacement_is_fenced_by_observed_etag() {
|
|||||||
assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner"));
|
assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn object_catalog_namespace_property_update_is_fenced_by_observed_etag() {
|
||||||
|
let backend = TestCatalogObjectBackend::default();
|
||||||
|
let store = ObjectTableCatalogStore::new(backend.clone());
|
||||||
|
let bucket = "analytics";
|
||||||
|
let namespace = Namespace::parse("sales").expect("namespace should parse");
|
||||||
|
store
|
||||||
|
.put_table_bucket(test_bucket_entry(bucket))
|
||||||
|
.await
|
||||||
|
.expect("table bucket should be created");
|
||||||
|
store
|
||||||
|
.create_namespace(test_namespace_entry(bucket, &namespace))
|
||||||
|
.await
|
||||||
|
.expect("namespace should be created");
|
||||||
|
|
||||||
|
let namespace_path = store.paths.namespace_entry_path(bucket, &namespace);
|
||||||
|
let pause = backend.pause_next_put(RUSTFS_META_BUCKET, &namespace_path).await;
|
||||||
|
let stale_store = store.clone();
|
||||||
|
let stale_update = tokio::spawn(async move {
|
||||||
|
stale_store
|
||||||
|
.update_namespace_properties(
|
||||||
|
bucket,
|
||||||
|
"sales",
|
||||||
|
NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "stale".to_string())]))
|
||||||
|
.expect("namespace update should validate"),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
});
|
||||||
|
pause.wait_started().await;
|
||||||
|
|
||||||
|
let mut winner = test_namespace_entry(bucket, &namespace);
|
||||||
|
winner.properties.insert("owner".to_string(), "winner".to_string());
|
||||||
|
backend
|
||||||
|
.seed_object(
|
||||||
|
RUSTFS_META_BUCKET,
|
||||||
|
&namespace_path,
|
||||||
|
serde_json::to_vec(&winner).expect("winning namespace should encode"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
pause.release();
|
||||||
|
|
||||||
|
assert_matches!(
|
||||||
|
stale_update.await.expect("stale namespace update task should finish"),
|
||||||
|
Err(TableCatalogStoreError::Conflict(_))
|
||||||
|
);
|
||||||
|
let stored = store
|
||||||
|
.get_namespace(bucket, &namespace.public_name())
|
||||||
|
.await
|
||||||
|
.expect("winning namespace should load")
|
||||||
|
.expect("winning namespace should remain");
|
||||||
|
assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner"));
|
||||||
|
}
|
||||||
|
|
||||||
async fn assert_direct_namespace_child_contract<S>(store: &S, bucket: &str, cursor_prefix: &str)
|
async fn assert_direct_namespace_child_contract<S>(store: &S, bucket: &str, cursor_prefix: &str)
|
||||||
where
|
where
|
||||||
S: TableCatalogStore + ?Sized,
|
S: TableCatalogStore + ?Sized,
|
||||||
|
|||||||
Reference in New Issue
Block a user