gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-13 16:46:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(table-catalog): add atomic table rename (#5989)

Browse Source 
Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
2026-08-12 21:39:18 +08:00
committed by GitHub
parent 2f83d6789b
commit c9eeb2fa8a
15 changed files with 917 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog/mod.rs
+89 -194
View File
@@ -76,6 +76,8 @@ const MIN_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60;
const MAX_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60 * 60;
const NAMESPACE_REQUEST_BODY_MAX_SIZE: usize = MAX_ADMIN_REQUEST_BODY_SIZE;
const NAMESPACE_REQUEST_BODY_TIMEOUT: StdDuration = StdDuration::from_secs(10);
const RENAME_TABLE_BODY_MAX_SIZE: usize = 16 * 1024;
const RENAME_TABLE_BODY_TIMEOUT: StdDuration = StdDuration::from_secs(10);
const WAREHOUSE_PROPERTY: &str = "warehouse";
const PREFIX_PROPERTY: &str = "prefix";
@@ -204,7 +206,10 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
"POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery",
"POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback",
];
const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &["POST /v1/{prefix}/namespaces/{namespace}/properties"];
const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &[
"POST /v1/{prefix}/namespaces/{namespace}/properties",
"POST /v1/{prefix}/tables/rename",
];
static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {};
@@ -229,6 +234,7 @@ static TABLE_EXISTS_HANDLER: RestTableExistsHandler = RestTableExistsHandler {};
static LOAD_CREDENTIALS_HANDLER: RestLoadCredentialsHandler = RestLoadCredentialsHandler {};
static COMMIT_TABLE_HANDLER: RestCommitTableHandler = RestCommitTableHandler {};
static DROP_TABLE_HANDLER: RestDropTableHandler = RestDropTableHandler {};
static RENAME_TABLE_HANDLER: RestRenameTableHandler = RestRenameTableHandler {};
static LOAD_VIEW_HANDLER: RestLoadViewHandler = RestLoadViewHandler {};
static VIEW_EXISTS_HANDLER: RestViewExistsHandler = RestViewExistsHandler {};
static REPLACE_VIEW_HANDLER: RestReplaceViewHandler = RestReplaceViewHandler {};
@@ -660,12 +666,20 @@ struct RestListNamespacesResponse {
next_page_token: Option<String>,
}
#[derive(Debug, Serialize)]
#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct RestTableIdentifier {
namespace: Vec<String>,
name: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct RenameTableRequest {
source: RestTableIdentifier,
destination: RestTableIdentifier,
}
#[derive(Debug, Serialize)]
struct RestListTablesResponse {
identifiers: Vec<RestTableIdentifier>,
@@ -1115,6 +1129,16 @@ async fn authorize_table_catalog_resource_request(
) -> S3Result<TableCatalogRequestPrincipal> {
let principal = table_catalog_request_principal(req).await?;
authorize_table_catalog_resource_for_principal(req, &principal, resource, action).await?;
Ok(principal)
}
async fn authorize_table_catalog_resource_for_principal(
req: &S3Request<Body>,
principal: &TableCatalogRequestPrincipal,
resource: &TableCatalogResource<'_>,
action: AdminAction,
) -> S3Result<()> {
let object_path = resource.object_path();
validate_admin_action_with_bucket_object_for_iam(
principal.iam_store.clone(),
@@ -1125,8 +1149,7 @@ async fn authorize_table_catalog_resource_request(
req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
AdminResourceScope::bucket_object(resource.warehouse, object_path.as_deref().unwrap_or("")),
)
.await?;
Ok(principal)
.await
}
struct TableCatalogRequestPrincipal {
@@ -3699,8 +3722,6 @@ struct SnapshotFileIdentity {
async fn validate_table_snapshot_commit_conflicts<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
current_metadata: &serde_json::Value,
updates: &[serde_json::Value],
@@ -3725,13 +3746,10 @@ where
.and_then(serde_json::Value::as_str)
.ok_or_else(|| s3_error!(InvalidRequest, "snapshot summary.operation is required"))?;
let current_live_files =
load_current_snapshot_live_files(metadata_backend, bucket, namespace, table, entry, current_metadata).await?;
let current_live_files = load_current_snapshot_live_files(metadata_backend, bucket, entry, current_metadata).await?;
let changes = load_snapshot_file_changes(
metadata_backend,
bucket,
namespace,
table,
entry,
snapshot,
SnapshotChangeContext {
@@ -3808,8 +3826,6 @@ fn added_snapshot_update(updates: &[serde_json::Value]) -> S3Result<Option<&serd
async fn load_current_snapshot_live_files<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
current_metadata: &serde_json::Value,
) -> S3Result<SnapshotLiveFiles>
@@ -3833,7 +3849,7 @@ where
.ok_or_else(|| s3_error!(InvalidRequest, "current snapshot metadata is missing"))?;
let mut live_files = SnapshotLiveFiles::default();
for manifest in read_snapshot_manifest_references(metadata_backend, bucket, namespace, table, entry, snapshot).await? {
for manifest in read_snapshot_manifest_references(metadata_backend, bucket, entry, snapshot).await? {
let SnapshotManifestLocation {
manifest_path,
sequence_number,
@@ -3877,8 +3893,6 @@ where
async fn load_snapshot_file_changes<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
snapshot: &serde_json::Value,
context: SnapshotChangeContext<'_>,
@@ -3887,7 +3901,7 @@ where
B: crate::table_catalog::TableCatalogObjectBackend,
{
let mut changes = SnapshotFileChanges::default();
for manifest in read_snapshot_manifest_references(metadata_backend, bucket, namespace, table, entry, snapshot).await? {
for manifest in read_snapshot_manifest_references(metadata_backend, bucket, entry, snapshot).await? {
let inherited_identity = context
.current_live_files
.manifest_files
@@ -3987,21 +4001,17 @@ struct SnapshotManifestReferences {
async fn read_snapshot_manifest_references<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
snapshot: &serde_json::Value,
) -> S3Result<Vec<SnapshotManifestReferences>>
where
B: crate::table_catalog::TableCatalogObjectBackend,
{
let manifest_locations = snapshot_manifest_locations(metadata_backend, bucket, namespace, table, entry, snapshot).await?;
let manifest_locations = snapshot_manifest_locations(metadata_backend, bucket, entry, snapshot).await?;
let mut manifests = Vec::new();
for manifest_location in manifest_locations {
let manifest_key = table_commit_object_key(
bucket,
namespace,
table,
entry,
&manifest_location.manifest_path,
crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestFile,
@@ -4026,7 +4036,7 @@ where
if reference.file_sequence_number.is_none() {
reference.file_sequence_number = manifest_location.sequence_number;
}
validate_manifest_data_file_reference(metadata_backend, bucket, namespace, table, entry, &reference).await?;
validate_manifest_data_file_reference(metadata_backend, bucket, entry, &reference).await?;
references.push(reference);
}
manifests.push(SnapshotManifestReferences {
@@ -4047,8 +4057,6 @@ struct SnapshotManifestLocation {
async fn snapshot_manifest_locations<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
snapshot: &serde_json::Value,
) -> S3Result<Vec<SnapshotManifestLocation>>
@@ -4058,8 +4066,6 @@ where
if let Some(manifest_list_location) = snapshot.get("manifest-list").and_then(serde_json::Value::as_str) {
let manifest_list_key = table_commit_object_key(
bucket,
namespace,
table,
entry,
manifest_list_location,
crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestList,
@@ -4105,15 +4111,13 @@ where
async fn validate_manifest_data_file_reference<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
reference: &crate::table_catalog::ManifestDataFileReference,
) -> S3Result<()>
where
B: crate::table_catalog::TableCatalogObjectBackend,
{
table_commit_object_key(bucket, namespace, table, entry, &reference.location, reference.object_kind.clone())?;
table_commit_object_key(bucket, entry, &reference.location, reference.object_kind.clone())?;
let object_key = crate::table_catalog::table_catalog_object_key_from_location(bucket, &reference.location)
.ok_or_else(|| s3_error!(InvalidRequest, "manifest data file location is invalid"))?;
if !metadata_backend
@@ -4128,8 +4132,6 @@ where
fn table_commit_object_key(
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
location: &str,
expected_kind: crate::table_catalog::TableMetadataMaintenanceObjectKind,
@@ -4138,7 +4140,7 @@ fn table_commit_object_key(
.ok_or_else(|| s3_error!(InvalidRequest, "snapshot object location is invalid"))?;
let warehouse_object_prefix = crate::table_catalog::table_warehouse_object_prefix(entry).map_err(catalog_store_error)?;
let object_kind =
crate::table_catalog::table_maintenance_object_kind(namespace, table, Some(&warehouse_object_prefix), &object_key)
crate::table_catalog::table_maintenance_object_kind_for_entry(entry, Some(&warehouse_object_prefix), &object_key)
.ok_or_else(|| s3_error!(InvalidRequest, "snapshot object is outside the table warehouse"))?;
if !crate::table_catalog::table_maintenance_object_kind_matches_reference(&object_kind, &expected_kind) {
return Err(s3_error!(InvalidRequest, "snapshot object kind does not match manifest metadata"));
@@ -4339,6 +4341,15 @@ fn catalog_store_error(err: crate::table_catalog::TableCatalogStoreError) -> S3E
crate::table_catalog::TableCatalogStoreError::NotFound(message) => {
iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_RESOURCE, StatusCode::NOT_FOUND, message)
}
crate::table_catalog::TableCatalogStoreError::NamespaceNotFound(message) => {
iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_NAMESPACE, StatusCode::NOT_FOUND, message)
}
crate::table_catalog::TableCatalogStoreError::TableNotFound(message) => {
iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, message)
}
crate::table_catalog::TableCatalogStoreError::AlreadyExists(message) => {
iceberg_rest_error(ICEBERG_ERROR_ALREADY_EXISTS, StatusCode::CONFLICT, message)
}
crate::table_catalog::TableCatalogStoreError::Conflict(message) => {
iceberg_rest_error(ICEBERG_ERROR_COMMIT_FAILED, StatusCode::CONFLICT, message)
}
@@ -4354,6 +4365,15 @@ fn catalog_store_error(err: crate::table_catalog::TableCatalogStoreError) -> S3E
}
}
fn table_identifier_from_request(
identifier: RestTableIdentifier,
) -> S3Result<(crate::table_catalog::Namespace, crate::table_catalog::IdentifierSegment)> {
let namespace = namespace_from_segments(&identifier.namespace)?;
let table = crate::table_catalog::IdentifierSegment::parse(identifier.name)
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
Ok((namespace, table))
}
fn catalog_store_conflict_error(err: crate::table_catalog::TableCatalogStoreError, conflict_type: &'static str) -> S3Error {
match err {
crate::table_catalog::TableCatalogStoreError::Conflict(message) => {
@@ -4547,12 +4567,10 @@ where
{
let mut entry = table_entry_from_register_request(bucket, namespace, request)?;
ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?;
let table = crate::table_catalog::IdentifierSegment::parse(entry.table.clone())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?;
validate_metadata_table_location_in_bucket(bucket, &metadata)?;
adopt_registered_metadata_identity(&mut entry, &metadata)?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table, &entry, None, &metadata).await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?;
store
.register_table_with_publication(entry.clone(), metadata_backend)
.await
@@ -4648,8 +4666,6 @@ async fn read_table_metadata_json(
async fn validate_table_metadata_snapshot_graph<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
current_metadata: Option<&serde_json::Value>,
metadata: &serde_json::Value,
@@ -4657,7 +4673,7 @@ async fn validate_table_metadata_snapshot_graph<B>(
where
B: crate::table_catalog::TableCatalogObjectBackend,
{
validate_table_metadata_snapshot_graph_result(metadata_backend, bucket, namespace, table, entry, current_metadata, metadata)
validate_table_metadata_snapshot_graph_result(metadata_backend, bucket, entry, current_metadata, metadata)
.await
.map_err(catalog_store_error)
}
@@ -4665,8 +4681,6 @@ where
async fn validate_table_metadata_snapshot_graph_result<B>(
metadata_backend: &B,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &crate::table_catalog::IdentifierSegment,
entry: &crate::table_catalog::TableEntry,
current_metadata: Option<&serde_json::Value>,
metadata: &serde_json::Value,
@@ -4676,8 +4690,7 @@ where
{
let mut target_entry = entry.clone();
target_entry.warehouse_location = crate::table_catalog::table_metadata_location(metadata)?.to_string();
let context =
crate::table_catalog::TableSnapshotGraphValidationContext::new(metadata_backend, bucket, namespace, table, &target_entry);
let context = crate::table_catalog::TableSnapshotGraphValidationContext::new(metadata_backend, bucket, &target_entry);
crate::table_catalog::validate_table_snapshot_changes(&context, current_metadata, metadata).await
}
@@ -4965,8 +4978,6 @@ async fn update_table_metadata_location_response<S>(
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let Some(current) = store
.load_table(bucket, &namespace.public_name(), table)
.await
@@ -4975,7 +4986,7 @@ where
return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found"));
};
let metadata_location = table_metadata_location_for_catalog(bucket, &request.metadata_location)?;
if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &metadata_location) {
if !crate::table_catalog::is_valid_table_metadata_location_for_entry(&current, &metadata_location) {
return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory"));
}
let existing_commit = table_commit_for_retry_ids(
@@ -4995,16 +5006,8 @@ where
validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
let table_bucket_fence_required = table_warehouse_location_changes(&current, &target_metadata)?;
validate_metadata_matches_current_metadata(&previous_metadata, &target_metadata)?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&previous_metadata),
&target_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, Some(&previous_metadata), &target_metadata)
.await?;
let requirements = match existing_commit.as_ref() {
Some(existing_commit) => replay_commit_requirements(existing_commit, &[], &target_metadata)?,
None => Vec::new(),
@@ -5056,8 +5059,6 @@ where
return standard_commit_table_response(store, metadata_backend, bucket, namespace, table, request).await;
}
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let Some(current) = store
.load_table(bucket, &namespace.public_name(), table)
.await
@@ -5073,7 +5074,7 @@ where
}
let client_requirements = request.requirements.clone();
let mut request = table_commit_request_from_rest_request(bucket, namespace, table, request)?;
if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &request.new_metadata_location) {
if !crate::table_catalog::is_valid_table_metadata_location_for_entry(&current, &request.new_metadata_location) {
return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory"));
}
let current_metadata = read_table_metadata_json(metadata_backend, bucket, &current.metadata_location).await?;
@@ -5095,41 +5096,16 @@ where
validate_metadata_table_location_in_bucket(bucket, &previous_metadata)?;
validate_table_commit_requirements(&previous_metadata, &client_requirements)?;
validate_metadata_matches_current_metadata(&previous_metadata, &target_metadata)?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&previous_metadata),
&target_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, Some(&previous_metadata), &target_metadata)
.await?;
let committed_metadata_location = request.new_metadata_location.clone();
let result = publish_table_commit(store, metadata_backend, table_bucket_fence_required, request).await?;
return commit_table_replay_response(
metadata_backend,
bucket,
namespace,
table,
result,
&committed_metadata_location,
target_metadata,
)
.await;
return commit_table_replay_response(metadata_backend, bucket, result, &committed_metadata_location, target_metadata)
.await;
}
validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
validate_table_commit_requirements(&current_metadata, &client_requirements)?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&current_metadata),
&target_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, Some(&current_metadata), &target_metadata).await?;
let result = publish_table_commit(store, metadata_backend, table_bucket_fence_required, request).await?;
Ok(commit_table_response_from_result(result, target_metadata))
}
@@ -5145,8 +5121,6 @@ async fn standard_commit_table_response<S>(
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let Some(current) = store
.load_table(bucket, &namespace.public_name(), table)
.await
@@ -5168,38 +5142,21 @@ where
apply_table_commit_updates_at(current_metadata, &request.updates, &previous_metadata_location, commit_timestamp_ms)?;
validate_metadata_table_location_in_bucket(bucket, &next_metadata)?;
validate_metadata_identity_matches_current_metadata(&expected_metadata, &next_metadata)?;
validate_table_metadata_snapshot_graph_result(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&expected_metadata),
&next_metadata,
)
.await
.map_err(|err| match err {
crate::table_catalog::TableCatalogStoreError::Invalid(message) => s3_error!(InvalidRequest, "{}", message),
err => catalog_store_error(err),
})?;
validate_table_snapshot_commit_conflicts(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
&expected_metadata,
&request.updates,
)
.await?;
validate_table_metadata_snapshot_graph_result(metadata_backend, bucket, &current, Some(&expected_metadata), &next_metadata)
.await
.map_err(|err| match err {
crate::table_catalog::TableCatalogStoreError::Invalid(message) => s3_error!(InvalidRequest, "{}", message),
err => catalog_store_error(err),
})?;
validate_table_snapshot_commit_conflicts(metadata_backend, bucket, &current, &expected_metadata, &request.updates).await?;
validate_metadata_matches_current_metadata(&expected_metadata, &next_metadata)?;
let (commit_id, metadata_file_token) = standard_commit_ids(request.commit_id.or_else(|| request.idempotency_key.clone()));
let next_generation = current.generation.saturating_add(1);
let next_metadata_location = crate::table_catalog::default_table_metadata_file_path(
namespace,
&table_name,
let next_metadata_location = crate::table_catalog::table_metadata_file_path_for_entry(
&current,
&next_metadata_file_name(next_generation, &metadata_file_token),
);
)
.map_err(catalog_store_error)?;
let next_metadata_data = serde_json::to_vec(&next_metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize table metadata update: {}", err))?;
let put_result = metadata_backend
@@ -5334,8 +5291,6 @@ fn replay_commit_requirements(
async fn commit_table_replay_response(
metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &str,
result: crate::table_catalog::TableCommitResult,
committed_metadata_location: &str,
committed_metadata: serde_json::Value,
@@ -5343,9 +5298,7 @@ async fn commit_table_replay_response(
let metadata = if result.table.metadata_location == committed_metadata_location {
committed_metadata
} else {
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &result.table.metadata_location) {
if !crate::table_catalog::is_valid_table_metadata_location_for_entry(&result.table, &result.table.metadata_location) {
return Err(iceberg_rest_error(
ICEBERG_ERROR_REST,
StatusCode::INTERNAL_SERVER_ERROR,
@@ -5421,13 +5374,9 @@ where
));
}
if crate::table_catalog::table_matches_staged_base(current, &commit) {
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
validate_table_metadata_snapshot_graph_result(
metadata_backend,
bucket,
namespace,
&table_name,
current,
Some(&previous_metadata),
&target_metadata,
@@ -5437,16 +5386,7 @@ where
crate::table_catalog::TableCatalogStoreError::Invalid(message) => s3_error!(InvalidRequest, "{}", message),
err => catalog_store_error(err),
})?;
validate_table_snapshot_commit_conflicts(
metadata_backend,
bucket,
namespace,
&table_name,
current,
&previous_metadata,
&request.updates,
)
.await?;
validate_table_snapshot_commit_conflicts(metadata_backend, bucket, current, &previous_metadata, &request.updates).await?;
}
let requirements = replay_commit_requirements(&commit, &request.requirements, &target_metadata)?;
let committed_metadata_location = commit.new_metadata_location.clone();
@@ -5471,16 +5411,7 @@ where
)
.await?;
Ok(Some(
commit_table_replay_response(
metadata_backend,
bucket,
namespace,
table,
result,
&committed_metadata_location,
target_metadata,
)
.await?,
commit_table_replay_response(metadata_backend, bucket, result, &committed_metadata_location, target_metadata).await?,
))
}
@@ -5682,25 +5613,14 @@ where
let next_metadata = apply_table_commit_updates(current_metadata.clone(), &updates, &previous_metadata_location)?;
validate_metadata_matches_current_metadata(&current_metadata, &next_metadata)?;
validate_metadata_table_location_in_bucket(bucket, &next_metadata)?;
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&current_metadata),
&next_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, Some(&current_metadata), &next_metadata).await?;
let (commit_id, metadata_file_token) = standard_commit_ids(None);
let next_generation = current.generation.saturating_add(1);
let next_metadata_location = crate::table_catalog::default_table_metadata_file_path(
namespace,
&table_name,
let next_metadata_location = crate::table_catalog::table_metadata_file_path_for_entry(
&current,
&next_metadata_file_name(next_generation, &metadata_file_token),
);
)
.map_err(catalog_store_error)?;
let next_metadata_data = serde_json::to_vec(&next_metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize snapshot expiration metadata: {}", err))?;
metadata_backend
@@ -6187,9 +6107,6 @@ where
let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?;
validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
let external_table_uuid = validate_external_catalog_metadata_uuid(request.external_table_uuid.as_deref(), &target_metadata)?;
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let (action, table_response) = if let Some(current) = store
.load_table(bucket, &namespace.public_name(), table)
.await
@@ -6206,16 +6123,8 @@ where
let current_metadata = read_table_metadata_json(metadata_backend, bucket, &current.metadata_location).await?;
validate_metadata_table_location_in_bucket(bucket, &current_metadata)?;
validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
Some(&current_metadata),
&target_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, Some(&current_metadata), &target_metadata)
.await?;
let table_bucket_fence_required = table_warehouse_location_changes(&current, &target_metadata)?;
let result = publish_table_commit(
store,
@@ -6257,8 +6166,7 @@ where
},
)?;
adopt_registered_metadata_identity(&mut entry, &target_metadata)?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table_name, &entry, None, &target_metadata)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &target_metadata).await?;
store
.register_table_with_publication(entry.clone(), metadata_backend)
.await
@@ -6298,12 +6206,10 @@ where
let result = async {
ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?;
let mut entry = table_entry_from_import_request(bucket, namespace, table, request)?;
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?;
validate_metadata_table_location_in_bucket(bucket, &metadata)?;
adopt_registered_metadata_identity(&mut entry, &metadata)?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table_name, &entry, None, &metadata).await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?;
if let Some(existing) = store
.load_table(bucket, &namespace.public_name(), table)
.await
@@ -6351,10 +6257,8 @@ where
else {
return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found"));
};
let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
.map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
let metadata_location = table_metadata_location_for_catalog(bucket, &request.metadata_location)?;
if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &metadata_location) {
if !crate::table_catalog::is_valid_table_metadata_location_for_entry(&current, &metadata_location) {
return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory"));
}
let current_metadata = read_table_metadata_json(metadata_backend, bucket, &current.metadata_location).await?;
@@ -6362,16 +6266,7 @@ where
let target_metadata = read_table_metadata_json(metadata_backend, bucket, &metadata_location).await?;
validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
validate_table_metadata_snapshot_graph(
metadata_backend,
bucket,
namespace,
&table_name,
&current,
None,
&target_metadata,
)
.await?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &current, None, &target_metadata).await?;
let table_bucket_fence_required = table_warehouse_location_changes(&current, &target_metadata)?;
let commit_request = crate::table_catalog::TableCommitRequest {
table_bucket: bucket.to_string(),
rustfs/src/admin/handlers/table_catalog/routes.rs
+5
View File
@@ -79,6 +79,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
AdminOperation(&DROP_NAMESPACE_HANDLER),
)?;
r.insert(
Method::POST,
format!("{prefix}/{{warehouse}}/tables/rename").as_str(),
AdminOperation(&RENAME_TABLE_HANDLER),
)?;
r.insert(
Method::GET,
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(),
rustfs/src/admin/handlers/table_catalog/table.rs
+40
View File
@@ -30,6 +30,46 @@ impl Operation for RestListTablesHandler {
}
}
pub struct RestRenameTableHandler {}
#[async_trait::async_trait]
impl Operation for RestRenameTableHandler {
async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let principal = table_catalog_request_principal(&req).await?;
let request = read_bounded_json_body::<RenameTableRequest>(
&req.headers,
std::mem::take(&mut req.input),
RENAME_TABLE_BODY_MAX_SIZE,
RENAME_TABLE_BODY_TIMEOUT,
"rename table",
)
.await?;
let (source_namespace, source_table) = table_identifier_from_request(request.source)?;
let (destination_namespace, destination_table) = table_identifier_from_request(request.destination)?;
let source_resource = TableCatalogResource::table(&warehouse, &source_namespace, source_table.as_str());
authorize_table_catalog_resource_for_principal(&req, &principal, &source_resource, AdminAction::SetTableAction).await?;
let destination_resource = TableCatalogResource::table(&warehouse, &destination_namespace, destination_table.as_str());
authorize_table_catalog_resource_for_principal(&req, &principal, &destination_resource, AdminAction::SetTableAction)
.await?;
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let store = table_catalog_store_from_extensions(&req.extensions)?;
store
.rename_table(
&warehouse,
&source_namespace.public_name(),
source_table.as_str(),
&destination_namespace.public_name(),
destination_table.as_str(),
)
.await
.map_err(catalog_store_error)?;
Ok(empty_response(StatusCode::NO_CONTENT))
}
}
pub struct RestCreateTableHandler {}
#[async_trait::async_trait]
rustfs/src/admin/handlers/table_catalog/tests.rs
+147
View File
@@ -155,6 +155,7 @@ fn catalog_config_response_lists_standard_rest_endpoints() {
.endpoints
.contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties")
);
assert!(!response.endpoints.contains(&"POST /v1/{prefix}/tables/rename"));
assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities");
assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot");
assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog");
@@ -279,6 +280,7 @@ fn catalog_config_response_reports_durable_strong_backing_override() {
.endpoints
.contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties")
);
assert!(response.endpoints.contains(&"POST /v1/{prefix}/tables/rename"));
}
#[test]
@@ -329,6 +331,28 @@ fn catalog_conflicts_use_operation_specific_iceberg_errors() {
));
assert_eq!(unsupported.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into()));
assert_eq!(unsupported.status_code(), Some(StatusCode::NOT_ACCEPTABLE));
for (error, expected_code, expected_status) in [
(
crate::table_catalog::TableCatalogStoreError::NamespaceNotFound("namespace not found".to_string()),
ICEBERG_ERROR_NO_SUCH_NAMESPACE,
StatusCode::NOT_FOUND,
),
(
crate::table_catalog::TableCatalogStoreError::TableNotFound("table not found".to_string()),
ICEBERG_ERROR_NO_SUCH_TABLE,
StatusCode::NOT_FOUND,
),
(
crate::table_catalog::TableCatalogStoreError::AlreadyExists("destination exists".to_string()),
ICEBERG_ERROR_ALREADY_EXISTS,
StatusCode::CONFLICT,
),
] {
let mapped = catalog_store_error(error);
assert_eq!(mapped.code(), &S3ErrorCode::Custom(expected_code.into()));
assert_eq!(mapped.status_code(), Some(expected_status));
}
}
#[test]
@@ -423,6 +447,16 @@ fn table_catalog_handlers_require_table_admin_actions() {
"external catalog sync should branch authorization on current table existence"
);
let rename_block = operation_block(&src, "RestRenameTableHandler");
assert_eq!(rename_block.matches("table_catalog_request_principal(&req).await?;").count(), 1);
assert_eq!(
rename_block
.matches("authorize_table_catalog_resource_for_principal(")
.count(),
2
);
assert_eq!(rename_block.matches("AdminAction::SetTableAction").count(), 2);
let migration_block = operation_block(&src, "GetTableCatalogMigrationHandler");
assert!(
migration_block.contains("TableCatalogResource::warehouse(&warehouse)"),
@@ -572,6 +606,7 @@ fn table_catalog_handlers_require_enabled_table_bucket_marker_before_catalog_sta
"RestNamespaceExistsHandler",
"RestUpdateNamespacePropertiesHandler",
"RestDropNamespaceHandler",
"RestRenameTableHandler",
"RestListTablesHandler",
"RestCreateTableHandler",
"RestRegisterTableHandler",
@@ -717,6 +752,7 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() {
let _: &RestLoadCredentialsHandler = &LOAD_CREDENTIALS_HANDLER;
let _: &RestCommitTableHandler = &COMMIT_TABLE_HANDLER;
let _: &RestDropTableHandler = &DROP_TABLE_HANDLER;
let _: &RestRenameTableHandler = &RENAME_TABLE_HANDLER;
let _: &RestLoadViewHandler = &LOAD_VIEW_HANDLER;
let _: &RestReplaceViewHandler = &REPLACE_VIEW_HANDLER;
let _: &RestDropViewHandler = &DROP_VIEW_HANDLER;
@@ -760,6 +796,7 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() {
assert_operation::<RestLoadCredentialsHandler>();
assert_operation::<RestCommitTableHandler>();
assert_operation::<RestDropTableHandler>();
assert_operation::<RestRenameTableHandler>();
assert_operation::<RestLoadViewHandler>();
assert_operation::<RestReplaceViewHandler>();
assert_operation::<RestDropViewHandler>();
@@ -940,6 +977,14 @@ fn table_catalog_ingress_requests_reject_unknown_fields() {
"unexpected": true
}),
);
assert_rejects_unknown_field::<RenameTableRequest>(
"RenameTableRequest",
serde_json::json!({
"source": {"namespace": ["analytics"], "name": "events"},
"destination": {"namespace": ["curated"], "name": "events_v2"},
"unexpected": true
}),
);
assert_rejects_unknown_field::<RegisterTableRequest>(
"RegisterTableRequest",
serde_json::json!({
@@ -1020,6 +1065,57 @@ fn table_catalog_ingress_requests_reject_unknown_fields() {
);
}
#[test]
fn rename_table_request_uses_standard_identifiers_and_strict_serde() {
let request: RenameTableRequest = serde_json::from_value(serde_json::json!({
"source": {"namespace": ["analytics", "raw"], "name": "events"},
"destination": {"namespace": ["analytics", "curated"], "name": "events_v2"}
}))
.expect("rename request should parse");
assert_eq!(request.source.namespace, vec!["analytics", "raw"]);
assert_eq!(request.source.name, "events");
assert_eq!(request.destination.namespace, vec!["analytics", "curated"]);
assert_eq!(request.destination.name, "events_v2");
assert_rejects_unknown_field::<RenameTableRequest>(
"RenameTableRequest.source",
serde_json::json!({
"source": {"namespace": ["analytics"], "name": "events", "unexpected": true},
"destination": {"namespace": ["curated"], "name": "events_v2"}
}),
);
}
#[tokio::test]
async fn rename_table_body_rejects_declared_and_streamed_oversize_payloads() {
let mut oversized_headers = HeaderMap::new();
oversized_headers.insert(
http::header::CONTENT_LENGTH,
HeaderValue::from_str(&(RENAME_TABLE_BODY_MAX_SIZE + 1).to_string()).expect("content length should parse"),
);
let declared = read_bounded_json_body::<RenameTableRequest>(
&oversized_headers,
Body::empty(),
RENAME_TABLE_BODY_MAX_SIZE,
RENAME_TABLE_BODY_TIMEOUT,
"rename table",
)
.await
.expect_err("oversized declared body should fail before reading");
assert_eq!(declared.code(), &S3ErrorCode::InvalidRequest);
let streamed = read_bounded_json_body::<RenameTableRequest>(
&HeaderMap::new(),
Body::from(vec![b' '; RENAME_TABLE_BODY_MAX_SIZE + 1]),
RENAME_TABLE_BODY_MAX_SIZE,
RENAME_TABLE_BODY_TIMEOUT,
"rename table",
)
.await
.expect_err("oversized streamed body should fail");
assert_eq!(streamed.code(), &S3ErrorCode::InvalidRequest);
}
fn assert_rejects_unknown_field<T>(target: &str, value: serde_json::Value)
where
T: serde::de::DeserializeOwned,
@@ -2381,6 +2477,57 @@ async fn standard_commit_applies_updates_and_writes_next_metadata() {
);
}
#[tokio::test]
async fn standard_commit_after_table_rename_keeps_the_original_metadata_root() {
let metadata_backend = TestTableCatalogObjectBackend::default();
let store = crate::table_catalog::StrongTableCatalogStore::new(metadata_backend.clone());
let source_namespace = crate::table_catalog::Namespace::parse("analytics").expect("source namespace should parse");
let destination_namespace = crate::table_catalog::Namespace::parse("curated").expect("destination namespace should parse");
let created = create_standard_events_table(&store, &metadata_backend, &source_namespace).await;
create_namespace_response(
&store,
"warehouse",
CreateNamespaceRequest {
namespace: vec!["curated".to_string()],
properties: BTreeMap::new(),
},
true,
)
.await
.expect("destination namespace should be created");
store
.rename_table("warehouse", "analytics", "events", "curated", "events_v2")
.await
.expect("table should rename");
let request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({
"requirements": [{"type": "assert-table-uuid", "uuid": created.metadata["table-uuid"]}],
"updates": [{"action": "set-properties", "updates": {"owner": "curated"}}]
}))
.expect("commit request should parse");
let committed = standard_commit_table_response(
&store,
&trusted_table_commit_backend(&metadata_backend),
"warehouse",
&destination_namespace,
"events_v2",
request,
)
.await
.expect("renamed table should accept a standard commit");
let original_metadata_root = crate::table_catalog::default_table_metadata_dir_path(
&source_namespace,
&crate::table_catalog::IdentifierSegment::parse("events").expect("source table should parse"),
);
assert!(
committed
.metadata_location
.starts_with(&format!("s3://warehouse/{original_metadata_root}/"))
);
assert!(!committed.metadata_location.contains("/namespaces/curated/tables/events_v2/"));
}
#[tokio::test]
async fn standard_commit_uses_client_uuid_commit_id_in_metadata_file_name() {
let store = TestTableCatalogStore::default();
rustfs/src/admin/route_policy.rs
+10 -1
View File
@@ -930,6 +930,7 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
DELETE_TABLE_NAMESPACE,
RouteRiskLevel::High,
),
admin(HttpMethod::Post, "/iceberg/v1/{warehouse}/tables/rename", SET_TABLE, RouteRiskLevel::High),
admin(
HttpMethod::Get,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables",
@@ -1213,6 +1214,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
DELETE_TABLE_NAMESPACE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/tables/rename",
SET_TABLE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables",
@@ -1660,7 +1667,7 @@ 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(), 96);
assert_eq!(table_specs.count(), 98);
assert_action(HttpMethod::Put, "/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);
@@ -1679,6 +1686,8 @@ mod tests {
);
assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE);
assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE);
assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/tables/rename", SET_TABLE);
assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/tables/rename", SET_TABLE);
assert_action(
HttpMethod::Get,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/views",
rustfs/src/admin/route_registration_test.rs
+4
View File
@@ -405,6 +405,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/register",
"/analytics/namespaces/sales/register",
),
table_route_sample(Method::POST, "/{warehouse}/tables/rename", "/analytics/tables/rename"),
table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/views",
@@ -601,6 +602,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/register",
"/analytics/namespaces/sales/register",
),
compat_table_route_sample(Method::POST, "/{warehouse}/tables/rename", "/analytics/tables/rename"),
compat_table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/views",
@@ -909,6 +911,7 @@ fn test_register_routes_cover_representative_admin_paths() {
assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables"));
assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables"));
assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/register"));
assert_route(&router, Method::POST, &table_catalog_path("/analytics/tables/rename"));
assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/views"));
assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/views"));
assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders"));
@@ -1059,6 +1062,7 @@ fn test_register_routes_cover_representative_admin_paths() {
assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/tables"));
assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/tables"));
assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/register"));
assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/tables/rename"));
assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/views"));
assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/views"));
assert_route(
rustfs/src/table_catalog/error.rs
+6
View File
@@ -58,6 +58,9 @@ impl std::error::Error for TableObjectMutationError {}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum TableCatalogStoreError {
NotFound(String),
NamespaceNotFound(String),
TableNotFound(String),
AlreadyExists(String),
Conflict(String),
Invalid(String),
Unsupported(String),
@@ -68,6 +71,9 @@ impl fmt::Display for TableCatalogStoreError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::NotFound(message) => write!(f, "table catalog entry not found: {message}"),
Self::NamespaceNotFound(message) => write!(f, "table catalog namespace not found: {message}"),
Self::TableNotFound(message) => write!(f, "table catalog table not found: {message}"),
Self::AlreadyExists(message) => write!(f, "table catalog entry already exists: {message}"),
Self::Conflict(message) => write!(f, "table catalog conflict: {message}"),
Self::Invalid(message) => write!(f, "invalid table catalog entry: {message}"),
Self::Unsupported(message) => write!(f, "unsupported table catalog operation: {message}"),
rustfs/src/table_catalog/iceberg/commit.rs
+6 -2
View File
@@ -269,9 +269,13 @@ pub(crate) fn record_table_commit_attempt(operation: &str) {
fn table_catalog_store_result_label<T>(result: &TableCatalogStoreResult<T>) -> &'static str {
match result {
Ok(_) => "success",
Err(TableCatalogStoreError::Conflict(_)) => "conflict",
Err(TableCatalogStoreError::Conflict(_) | TableCatalogStoreError::AlreadyExists(_)) => "conflict",
Err(TableCatalogStoreError::Invalid(_)) => "invalid",
Err(TableCatalogStoreError::NotFound(_)) => "not_found",
Err(
TableCatalogStoreError::NotFound(_)
| TableCatalogStoreError::NamespaceNotFound(_)
| TableCatalogStoreError::TableNotFound(_),
) => "not_found",
Err(TableCatalogStoreError::Unsupported(_)) => "unsupported",
Err(TableCatalogStoreError::Internal(_)) => "failure",
}
rustfs/src/table_catalog/iceberg/validation.rs
+3 -14
View File
@@ -859,24 +859,14 @@ fn max_partition_field_id(value: &serde_json::Value) -> i64 {
pub(crate) struct TableSnapshotGraphValidationContext<'a, B> {
backend: &'a B,
table_bucket: &'a str,
namespace: &'a Namespace,
table: &'a IdentifierSegment,
entry: &'a TableEntry,
}
impl<'a, B> TableSnapshotGraphValidationContext<'a, B> {
pub(crate) fn new(
backend: &'a B,
table_bucket: &'a str,
namespace: &'a Namespace,
table: &'a IdentifierSegment,
entry: &'a TableEntry,
) -> Self {
pub(crate) fn new(backend: &'a B, table_bucket: &'a str, entry: &'a TableEntry) -> Self {
Self {
backend,
table_bucket,
namespace,
table,
entry,
}
}
@@ -1440,9 +1430,8 @@ fn snapshot_graph_object_key<B>(
let object_key = table_catalog_object_key_from_location(context.table_bucket, location)
.ok_or_else(|| TableCatalogStoreError::Invalid("snapshot object location is invalid".to_string()))?;
let warehouse_object_prefix = table_warehouse_object_prefix(context.entry)?;
let object_kind =
table_maintenance_object_kind(context.namespace, context.table, Some(&warehouse_object_prefix), &object_key)
.ok_or_else(|| TableCatalogStoreError::Invalid("snapshot object is outside the table warehouse".to_string()))?;
let object_kind = table_maintenance_object_kind_for_entry(context.entry, Some(&warehouse_object_prefix), &object_key)
.ok_or_else(|| TableCatalogStoreError::Invalid("snapshot object is outside the table warehouse".to_string()))?;
if !table_maintenance_object_kind_matches_reference(&object_kind, &expected_kind) {
return Err(TableCatalogStoreError::Invalid(
"snapshot object kind does not match manifest metadata".to_string(),
rustfs/src/table_catalog/identifier.rs
+62
View File
@@ -268,6 +268,68 @@ pub(crate) fn metadata_location_from_metadata_file_path(
.map(|_| object_key.to_string())
}
fn table_metadata_dir_from_object_key(object_key: &str) -> Option<String> {
let namespace_root = default_namespace_root_prefix();
let relative = object_key.strip_prefix(&namespace_root)?;
let (namespace_storage_id, table_path) = relative.rsplit_once(&format!("/{TABLE_ROOT}/"))?;
Namespace::from_segments(namespace_storage_id.split('/').map(str::to_string).collect()).ok()?;
let (table_name, metadata_file_name) = table_path.split_once(&format!("/{METADATA_DIR}/"))?;
IdentifierSegment::parse(table_name).ok()?;
if !is_valid_table_metadata_file_name(metadata_file_name) {
return None;
}
Some(format!("{namespace_root}{namespace_storage_id}/{TABLE_ROOT}/{table_name}/{METADATA_DIR}"))
}
pub(crate) fn table_metadata_dir_path_for_entry(entry: &TableEntry) -> TableCatalogStoreResult<String> {
let object_key = table_catalog_object_key_from_location(&entry.table_bucket, &entry.metadata_location).ok_or_else(|| {
TableCatalogStoreError::Invalid("current metadata location must be inside a table metadata directory".to_string())
})?;
if let Some(metadata_dir) = table_metadata_dir_from_object_key(&object_key) {
return Ok(metadata_dir);
}
if is_reserved_table_object_key(&object_key) {
return Err(TableCatalogStoreError::Invalid(
"current metadata location has an invalid protected table metadata path".to_string(),
));
}
let (metadata_dir, metadata_file_name) = object_key.rsplit_once('/').ok_or_else(|| {
TableCatalogStoreError::Invalid("current metadata location must be inside a table metadata directory".to_string())
})?;
if metadata_dir
.strip_suffix(&format!("/{METADATA_DIR}"))
.is_none_or(str::is_empty)
|| !is_valid_table_metadata_file_name(metadata_file_name)
{
return Err(TableCatalogStoreError::Invalid(
"current metadata location must be inside a table metadata directory".to_string(),
));
}
Ok(metadata_dir.to_string())
}
pub(crate) fn is_valid_table_metadata_location_for_entry(entry: &TableEntry, metadata_location: &str) -> bool {
let Ok(metadata_dir) = table_metadata_dir_path_for_entry(entry) else {
return false;
};
let Some(object_key) = table_catalog_object_key_from_location(&entry.table_bucket, metadata_location) else {
return false;
};
object_key
.strip_prefix(&format!("{metadata_dir}/"))
.is_some_and(is_valid_table_metadata_file_name)
}
pub(crate) fn table_metadata_file_path_for_entry(
entry: &TableEntry,
metadata_file_name: &str,
) -> TableCatalogStoreResult<String> {
if !is_valid_table_metadata_file_name(metadata_file_name) {
return Err(TableCatalogStoreError::Invalid("invalid table metadata file name".to_string()));
}
Ok(format!("{}/{}", table_metadata_dir_path_for_entry(entry)?, metadata_file_name))
}
pub(crate) fn is_valid_table_metadata_location(
namespace: &Namespace,
table: &IdentifierSegment,
rustfs/src/table_catalog/maintenance/recovery.rs
+51
View File
@@ -471,6 +471,57 @@ pub(crate) fn table_maintenance_object_kind(
None
}
pub(crate) fn table_maintenance_object_kind_for_entry(
entry: &TableEntry,
warehouse_object_prefix: Option<&str>,
object_location: &str,
) -> Option<TableMetadataMaintenanceObjectKind> {
let metadata_dir = table_metadata_dir_path_for_entry(entry).ok()?;
let metadata_prefix = format!("{metadata_dir}/");
if let Some(kind) = table_maintenance_metadata_object_kind(&metadata_prefix, object_location) {
return Some(kind);
}
let table_root = metadata_dir.strip_suffix(&format!("/{METADATA_DIR}"))?;
let data_prefix = format!("{table_root}/{DATA_DIR}/");
if object_location
.strip_prefix(&data_prefix)
.is_some_and(is_valid_table_maintenance_nested_object)
{
return Some(TableMetadataMaintenanceObjectKind::DataFile);
}
let delete_prefix = format!("{table_root}/{DELETE_DIR}/");
if object_location
.strip_prefix(&delete_prefix)
.is_some_and(is_valid_table_maintenance_nested_object)
{
return Some(TableMetadataMaintenanceObjectKind::DeleteFile);
}
if let Some(warehouse_object_prefix) = warehouse_object_prefix {
let metadata_prefix = format!("{warehouse_object_prefix}{METADATA_DIR}/");
if let Some(kind) = table_maintenance_metadata_object_kind(&metadata_prefix, object_location) {
return Some(kind);
}
let data_prefix = format!("{warehouse_object_prefix}{DATA_DIR}/");
if object_location
.strip_prefix(&data_prefix)
.is_some_and(is_valid_table_maintenance_nested_object)
{
return Some(TableMetadataMaintenanceObjectKind::DataFile);
}
let delete_prefix = format!("{warehouse_object_prefix}{DELETE_DIR}/");
if object_location
.strip_prefix(&delete_prefix)
.is_some_and(is_valid_table_maintenance_nested_object)
{
return Some(TableMetadataMaintenanceObjectKind::DeleteFile);
}
}
None
}
pub(crate) fn table_maintenance_object_kind_matches_reference(
actual: &TableMetadataMaintenanceObjectKind,
referenced: &TableMetadataMaintenanceObjectKind,
rustfs/src/table_catalog/mod.rs
+3 -2
View File
@@ -74,8 +74,9 @@ pub use identifier::{IdentifierSegment, Namespace, is_reserved_table_object_key}
pub(crate) use identifier::{
default_table_bucket_publication_lock_path, default_table_data_dir_path, default_table_delete_dir_path,
default_table_metadata_dir_path, default_table_metadata_file_path, default_table_publication_lock_path,
default_view_metadata_file_path, is_valid_table_metadata_location, is_valid_view_metadata_location,
metadata_location_from_metadata_file_path, validate_bucket_object_mutation,
default_view_metadata_file_path, is_valid_table_metadata_location, is_valid_table_metadata_location_for_entry,
is_valid_view_metadata_location, metadata_location_from_metadata_file_path, table_metadata_dir_path_for_entry,
table_metadata_file_path_for_entry, validate_bucket_object_mutation,
};
pub(crate) use maintenance::*;
pub(crate) use model::*;
rustfs/src/table_catalog/store/mod.rs
+34
View File
@@ -224,6 +224,20 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<Option<TableEntry>>;
async fn rename_table(
&self,
table_bucket: &str,
source_namespace: &str,
source_table: &str,
destination_namespace: &str,
destination_table: &str,
) -> TableCatalogStoreResult<()> {
let _ = (table_bucket, source_namespace, source_table, destination_namespace, destination_table);
Err(TableCatalogStoreError::Unsupported(
"table rename is not supported by this catalog store".to_string(),
))
}
async fn resolve_table_data_plane_resource(
&self,
table_bucket: &str,
@@ -1092,6 +1106,26 @@ where
}
}
async fn rename_table(
&self,
table_bucket: &str,
source_namespace: &str,
source_table: &str,
destination_namespace: &str,
destination_table: &str,
) -> TableCatalogStoreResult<()> {
match self {
Self::ObjectBacked(_) => Err(TableCatalogStoreError::Unsupported(
"table rename requires durable-strong catalog backing".to_string(),
)),
Self::DurableStrong(store) => {
store
.rename_table(table_bucket, source_namespace, source_table, destination_namespace, destination_table)
.await
}
}
}
async fn resolve_table_data_plane_resource(
&self,
table_bucket: &str,
rustfs/src/table_catalog/store/strong.rs
+113 -9
View File
@@ -123,6 +123,11 @@ enum StrongSnapshotWritePostcondition {
key: StrongResourceKey,
table_id: String,
},
TableRenamed {
source_key: StrongResourceKey,
destination_key: StrongResourceKey,
table_id: String,
},
ViewPresent(ViewEntry),
ViewAbsent {
key: StrongResourceKey,
@@ -166,6 +171,20 @@ impl StrongSnapshotWritePostcondition {
== Some(expected)
}
Self::TableAbsent { key, table_id } => state.tables.get(key).is_none_or(|current| current.table_id != *table_id),
Self::TableRenamed {
source_key,
destination_key,
table_id,
} => {
state
.tables
.get(source_key)
.is_none_or(|current| current.table_id != *table_id)
&& state
.tables
.get(destination_key)
.is_some_and(|current| current.table_id == *table_id)
}
Self::ViewPresent(expected) => {
let (Ok(namespace), Ok(view)) =
(parse_namespace_for_store(&expected.namespace), parse_table_for_store(&expected.view))
@@ -813,9 +832,8 @@ where
continue;
}
let namespace_identity = parse_namespace_for_store(namespace)?;
let table_identity = parse_table_for_store(table)?;
validate_table_warehouse_location(table_bucket, &entry.warehouse_location)?;
if !is_valid_table_metadata_location(&namespace_identity, &table_identity, &entry.metadata_location) {
if !is_valid_table_metadata_location_for_entry(entry, &entry.metadata_location) {
return Err(TableCatalogStoreError::Invalid(format!(
"strong catalog table {table_bucket}/{namespace}/{table} has an invalid metadata location"
)));
@@ -909,9 +927,8 @@ where
continue;
}
let namespace_identity = parse_namespace_for_store(namespace)?;
let table_identity = parse_table_for_store(table)?;
validate_table_warehouse_location(table_bucket, &entry.warehouse_location)?;
if !is_valid_table_metadata_location(&namespace_identity, &table_identity, &entry.metadata_location) {
if !is_valid_table_metadata_location_for_entry(entry, &entry.metadata_location) {
return Err(TableCatalogStoreError::Invalid(format!(
"strong catalog table {table_bucket}/{namespace}/{table} has an invalid metadata location"
)));
@@ -1607,8 +1624,6 @@ where
state: &StrongTableCatalogState,
key: &StrongResourceKey,
request: &TableCommitRequest,
namespace: &Namespace,
table: &IdentifierSegment,
) -> TableCatalogStoreResult<TableEntry> {
Self::ensure_identifier_is_unambiguous_locked(state, key)?;
let Some(current) = state.tables.get(key).cloned() else {
@@ -1696,7 +1711,7 @@ where
"current table metadata location does not match expected location".to_string(),
));
}
if !is_valid_table_metadata_location(namespace, table, &request.new_metadata_location) {
if !is_valid_table_metadata_location_for_entry(&current, &request.new_metadata_location) {
return Err(TableCatalogStoreError::Invalid(
"new metadata location must be inside the table metadata directory".to_string(),
));
@@ -1763,7 +1778,7 @@ where
next_warehouse_location: Option<String>,
) -> TableCatalogStoreResult<TableCommitResult> {
let key = Self::table_key(&request.table_bucket, namespace, table);
let current = Self::validate_new_table_commit_locked(state, &key, request, namespace, table)?;
let current = Self::validate_new_table_commit_locked(state, &key, request)?;
if let Some((result, _)) = Self::committed_existing_result_locked(state, request, current.clone()) {
return Ok(result);
}
@@ -2264,6 +2279,95 @@ where
.cloned())
}
async fn rename_table(
&self,
table_bucket: &str,
source_namespace: &str,
source_table: &str,
destination_namespace: &str,
destination_table: &str,
) -> TableCatalogStoreResult<()> {
let publication = TableCommitLockPublication::new(&self.object_backend);
publication.begin_table_bucket(table_bucket).await?;
if !publication.holds_table_bucket(table_bucket) {
return Err(TableCatalogStoreError::Internal(
"table rename requires a table-bucket publication fence".to_string(),
));
}
let _publication_completion = TableCommitPublicationCompletion::new(&publication);
let _migration_guard = self.acquire_snapshot_write_permit().await?;
let _write_guard = self.write_lock.lock().await;
self.hydrate_state().await?;
let source_namespace = parse_namespace_for_store(source_namespace)?;
let source_table = parse_table_for_store(source_table)?;
let destination_namespace = parse_namespace_for_store(destination_namespace)?;
let destination_table = parse_table_for_store(destination_table)?;
let source_key = Self::table_key(table_bucket, &source_namespace, &source_table);
let destination_key = Self::table_key(table_bucket, &destination_namespace, &destination_table);
let (snapshot, precondition, postcondition) = {
let state = self.state.lock().await;
Self::require_table_bucket_in_state(&state, table_bucket)?;
Self::ensure_identifier_is_unambiguous_locked(&state, &source_key)?;
Self::ensure_identifier_is_unambiguous_locked(&state, &destination_key)?;
let source = state
.tables
.get(&source_key)
.filter(|entry| entry.state == TableCatalogEntryState::Active)
.cloned()
.ok_or_else(|| {
TableCatalogStoreError::TableNotFound(format!(
"{table_bucket}/{}/{}",
source_namespace.public_name(),
source_table.as_str()
))
})?;
if !Self::namespace_exists_locked(&state, table_bucket, &source_namespace) {
return Err(TableCatalogStoreError::TableNotFound(format!(
"{table_bucket}/{}/{}",
source_namespace.public_name(),
source_table.as_str()
)));
}
if !Self::namespace_exists_locked(&state, table_bucket, &destination_namespace) {
return Err(TableCatalogStoreError::NamespaceNotFound(format!(
"{table_bucket}/{}",
destination_namespace.public_name()
)));
}
if state.tables.contains_key(&destination_key) || state.views.contains_key(&destination_key) {
return Err(TableCatalogStoreError::AlreadyExists(format!(
"destination table already exists: {table_bucket}/{}/{}",
destination_namespace.public_name(),
destination_table.as_str()
)));
}
if !is_valid_table_metadata_location_for_entry(&source, &source.metadata_location) {
return Err(TableCatalogStoreError::Invalid(
"current metadata location must be inside the table metadata directory".to_string(),
));
}
let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state);
draft_state.tables.remove(&source_key);
let mut destination = source;
destination.namespace = destination_namespace.public_name();
destination.table = destination_table.as_str().to_string();
draft_state.tables.insert(destination_key.clone(), destination.clone());
(
Self::snapshot_from_mutated_state_locked(&mut draft_state, self.snapshot_write_version)?,
precondition,
StrongSnapshotWritePostcondition::TableRenamed {
source_key,
destination_key,
table_id: destination.table_id,
},
)
};
self.finalize_snapshot_write(snapshot, precondition, postcondition).await
}
async fn resolve_table_data_plane_resource(
&self,
table_bucket: &str,
@@ -2345,7 +2449,7 @@ where
let committed_existing_result = {
let state = self.state.lock().await;
let current = Self::validate_new_table_commit_locked(&state, &key, &request, &namespace, &table);
let current = Self::validate_new_table_commit_locked(&state, &key, &request);
match current {
Ok(current) => {
let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state);
rustfs/src/table_catalog/tests.rs
+344 -10
View File
@@ -1919,7 +1919,7 @@ async fn iceberg_snapshot_graph_rejects_unknown_partition_specs() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
let error = validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -1964,7 +1964,7 @@ async fn iceberg_snapshot_graph_allows_missing_deleted_files() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -1992,7 +1992,7 @@ async fn iceberg_snapshot_graph_accepts_empty_manifest_lists() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -2036,7 +2036,7 @@ async fn iceberg_v2_snapshot_graph_accepts_reused_v1_manifests() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -2080,7 +2080,7 @@ async fn iceberg_snapshot_change_validation_skips_unchanged_history() {
}));
next_metadata["current-snapshot-id"] = serde_json::Value::from(11);
next_metadata["refs"]["main"]["snapshot-id"] = serde_json::Value::from(11);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, Some(&current_metadata), &next_metadata)
.await
@@ -2116,7 +2116,7 @@ async fn iceberg_snapshot_registration_validates_only_active_snapshot_heads() {
]);
metadata["current-snapshot-id"] = serde_json::Value::from(11);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 11}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -2169,7 +2169,7 @@ async fn iceberg_snapshot_graph_counts_shared_manifests_once() {
metadata["current-snapshot-id"] = serde_json::Value::from(current_snapshot_id);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": current_snapshot_id}});
let current_metadata = table_metadata_json_for_validation();
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, Some(&current_metadata), &metadata)
.await
@@ -2227,7 +2227,7 @@ async fn iceberg_snapshot_graph_accepts_more_than_ten_thousand_live_files() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(20);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 20}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -2263,7 +2263,7 @@ async fn iceberg_v2_snapshot_graph_accepts_embedded_v2_manifests() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -2307,7 +2307,7 @@ async fn iceberg_snapshot_graph_accepts_delete_files_in_data_directory() {
}]);
metadata["current-snapshot-id"] = serde_json::Value::from(10);
metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &entry);
validate_table_snapshot_changes(&context, None, &metadata)
.await
@@ -16659,3 +16659,337 @@ fn resolver_builds_paths_under_reserved_table_boundary() {
".rustfs-table/warehouses/warehouse1/namespaces/analytics/daily/tables/events/metadata"
);
}
#[tokio::test]
async fn strong_catalog_table_rename_is_atomic_and_preserves_stable_table_state() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").expect("source namespace should parse");
let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse");
let source_table = IdentifierSegment::parse("orders").expect("source table should parse");
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("table bucket should be created");
store
.create_namespace(test_namespace_entry(bucket, &source_namespace))
.await
.expect("source namespace should be created");
store
.create_namespace(test_namespace_entry(bucket, &destination_namespace))
.await
.expect("destination namespace should be created");
let source = test_table_entry(
bucket,
&source_namespace,
&source_table,
default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"),
);
store
.create_table(source.clone())
.await
.expect("source table should be created");
store
.rename_table(bucket, "sales", "orders", "curated", "orders_v2")
.await
.expect("table should rename");
assert!(
store
.load_table(bucket, "sales", "orders")
.await
.expect("source lookup should succeed")
.is_none()
);
let destination = store
.load_table(bucket, "curated", "orders_v2")
.await
.expect("destination lookup should succeed")
.expect("renamed table should exist");
assert_eq!(destination.table_id, source.table_id);
assert_eq!(destination.table_uuid, source.table_uuid);
assert_eq!(destination.warehouse_location, source.warehouse_location);
assert_eq!(destination.metadata_location, source.metadata_location);
assert_eq!(destination.version_token, source.version_token);
assert_eq!(destination.generation, source.generation);
let old_manifest = format!(
"{}/manifest-00001.avro",
default_table_metadata_dir_path(&source_namespace, &source_table)
);
assert_eq!(
table_maintenance_object_kind_for_entry(&destination, None, &old_manifest),
Some(TableMetadataMaintenanceObjectKind::ManifestFile)
);
let resource = store
.resolve_table_data_plane_resource(bucket, "tables/table-id/data/part.parquet")
.await
.expect("data-plane lookup should succeed")
.expect("renamed table should own its warehouse prefix");
assert_eq!(resource.namespace, "curated");
assert_eq!(resource.table, "orders_v2");
let mut replacement = test_table_entry(
bucket,
&source_namespace,
&source_table,
default_table_metadata_file_path(&source_namespace, &source_table, "00001-replacement.metadata.json"),
);
replacement.table_id = "replacement-table-id".to_string();
replacement.table_uuid = "replacement-table-uuid".to_string();
replacement.warehouse_location = "s3://analytics/tables/replacement-table-id".to_string();
store
.create_table(replacement.clone())
.await
.expect("the source identifier should be reusable after rename");
let recreated_source = store
.load_table(bucket, "sales", "orders")
.await
.expect("recreated source lookup should succeed")
.expect("recreated source should exist");
assert_eq!(recreated_source.table_id, replacement.table_id);
assert_ne!(recreated_source.table_id, destination.table_id);
assert_ne!(recreated_source.table_uuid, destination.table_uuid);
assert_ne!(recreated_source.warehouse_location, destination.warehouse_location);
assert_ne!(recreated_source.metadata_location, destination.metadata_location);
let next_metadata_location =
table_metadata_file_path_for_entry(&destination, "00002.metadata.json").expect("next metadata path should resolve");
backend.seed_object(bucket, &next_metadata_location, b"{}".to_vec()).await;
let committed = store
.commit_table(TableCommitRequest {
table_bucket: bucket.to_string(),
namespace: "curated".to_string(),
table: "orders_v2".to_string(),
commit_id: "rename-followup-commit".to_string(),
idempotency_key: Some("rename-followup-commit".to_string()),
operation: "append".to_string(),
expected_version_token: destination.version_token,
expected_metadata_location: destination.metadata_location,
new_metadata_location: next_metadata_location.clone(),
requirements: Vec::new(),
writer: Some("rename-test".to_string()),
})
.await
.expect("renamed table should accept a commit in its stable metadata directory");
assert_eq!(committed.table.metadata_location, next_metadata_location);
let restarted = StrongTableCatalogStore::new(backend);
let restarted_source = restarted
.load_table(bucket, "sales", "orders")
.await
.expect("source lookup after restart should succeed")
.expect("recreated source should survive restart");
assert_eq!(restarted_source.table_id, replacement.table_id);
let restarted_destination = restarted
.load_table(bucket, "curated", "orders_v2")
.await
.expect("destination lookup after restart should succeed")
.expect("destination should survive restart");
assert_eq!(restarted_destination.metadata_location, committed.table.metadata_location);
assert_eq!(
table_metadata_file_path_for_entry(&restarted_destination, "00003.metadata.json")
.expect("stable metadata path should survive restart"),
default_table_metadata_file_path(&source_namespace, &source_table, "00003.metadata.json")
);
}
#[tokio::test]
async fn strong_catalog_table_rename_rejects_missing_and_conflicting_destinations() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend);
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").expect("source namespace should parse");
let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse");
let source_table = IdentifierSegment::parse("orders").expect("source table should parse");
let destination_table = IdentifierSegment::parse("orders_v2").expect("destination table should parse");
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("bucket should be created");
store
.create_namespace(test_namespace_entry(bucket, &source_namespace))
.await
.expect("source namespace should be created");
store
.create_namespace(test_namespace_entry(bucket, &destination_namespace))
.await
.expect("destination namespace should be created");
store
.create_table(test_table_entry(
bucket,
&source_namespace,
&source_table,
default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"),
))
.await
.expect("source table should be created");
let mut existing = test_table_entry(
bucket,
&destination_namespace,
&destination_table,
default_table_metadata_file_path(&destination_namespace, &destination_table, "00001.metadata.json"),
);
existing.table_id = "destination-table-id".to_string();
existing.table_uuid = "destination-table-uuid".to_string();
existing.warehouse_location = "s3://analytics/tables/destination-table-id".to_string();
store
.create_table(existing)
.await
.expect("destination table should be created");
assert_matches!(
store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await,
Err(TableCatalogStoreError::AlreadyExists(_))
);
assert_matches!(
store.rename_table(bucket, "sales", "orders", "missing", "orders_v3").await,
Err(TableCatalogStoreError::NamespaceNotFound(_))
);
assert_matches!(
store.rename_table(bucket, "sales", "missing", "curated", "orders_v3").await,
Err(TableCatalogStoreError::TableNotFound(_))
);
let destination_view = IdentifierSegment::parse("orders_view").expect("view should parse");
store
.create_view(test_view_entry(
bucket,
&destination_namespace,
&destination_view,
default_view_metadata_file_path(&destination_namespace, &destination_view, "00001.metadata.json"),
))
.await
.expect("destination view should be created");
assert_matches!(
store.rename_table(bucket, "sales", "orders", "curated", "orders_view").await,
Err(TableCatalogStoreError::AlreadyExists(_))
);
assert!(
store
.load_table(bucket, "sales", "orders")
.await
.expect("source lookup should succeed")
.is_some()
);
}
#[tokio::test]
async fn strong_catalog_table_rename_does_not_publish_failed_snapshot() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").expect("source namespace should parse");
let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse");
let source_table = IdentifierSegment::parse("orders").expect("source table should parse");
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("bucket should be created");
store
.create_namespace(test_namespace_entry(bucket, &source_namespace))
.await
.expect("source namespace should be created");
store
.create_namespace(test_namespace_entry(bucket, &destination_namespace))
.await
.expect("destination namespace should be created");
store
.create_table(test_table_entry(
bucket,
&source_namespace,
&source_table,
default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"),
))
.await
.expect("source table should be created");
backend
.fail_next_put(
RUSTFS_META_BUCKET,
&StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path(),
)
.await;
assert_matches!(
store.rename_table(bucket, "sales", "orders", "curated", "orders_v2").await,
Err(TableCatalogStoreError::Internal(_))
);
assert!(
store
.load_table(bucket, "sales", "orders")
.await
.expect("source lookup should succeed")
.is_some()
);
assert!(
store
.load_table(bucket, "curated", "orders_v2")
.await
.expect("destination lookup should succeed")
.is_none()
);
}
#[tokio::test]
async fn strong_catalog_table_rename_returns_success_after_committed_snapshot_reload_failure() {
let backend = TestCatalogObjectBackend::default();
let store = StrongTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let source_namespace = Namespace::parse("sales").expect("source namespace should parse");
let destination_namespace = Namespace::parse("curated").expect("destination namespace should parse");
let source_table = IdentifierSegment::parse("orders").expect("source table should parse");
store
.put_table_bucket(test_bucket_entry(bucket))
.await
.expect("bucket should be created");
store
.create_namespace(test_namespace_entry(bucket, &source_namespace))
.await
.expect("source namespace should be created");
store
.create_namespace(test_namespace_entry(bucket, &destination_namespace))
.await
.expect("destination namespace should be created");
store
.create_table(test_table_entry(
bucket,
&source_namespace,
&source_table,
default_table_metadata_file_path(&source_namespace, &source_table, "00001.metadata.json"),
))
.await
.expect("source table should be created");
backend
.fail_next_read(
RUSTFS_META_BUCKET,
&StrongTableCatalogStore::<TestCatalogObjectBackend>::snapshot_object_path(),
)
.await;
store
.rename_table(bucket, "sales", "orders", "curated", "orders_v2")
.await
.expect("durably committed rename should succeed despite local reload failure");
assert!(!store.is_hydrated_for_test().await);
assert!(
store
.load_table(bucket, "curated", "orders_v2")
.await
.expect("destination lookup should reload durable state")
.is_some()
);
}
#[tokio::test]
async fn configured_object_catalog_rejects_table_rename() {
let store =
ConfiguredTableCatalogStore::new_for_test(TestCatalogObjectBackend::default(), TableCatalogBackingMode::ObjectBacked);
assert_matches!(
store
.rename_table("analytics", "sales", "orders", "curated", "orders_v2")
.await,
Err(TableCatalogStoreError::Unsupported(_))
);
}