diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 94ed44b56..c07a66c25 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -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, } -#[derive(Debug, Serialize)] +#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] struct RestTableIdentifier { namespace: Vec, name: String, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct RenameTableRequest { + source: RestTableIdentifier, + destination: RestTableIdentifier, +} + #[derive(Debug, Serialize)] struct RestListTablesResponse { identifiers: Vec, @@ -1115,6 +1129,16 @@ async fn authorize_table_catalog_resource_request( ) -> S3Result { 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, + 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::>().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( 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( 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 @@ -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( 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( 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> 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( 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> @@ -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( 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( 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( 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( 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( 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(¤t, &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(¤t, &target_metadata)?; validate_metadata_matches_current_metadata(&previous_metadata, &target_metadata)?; - validate_table_metadata_snapshot_graph( - metadata_backend, - bucket, - namespace, - &table_name, - ¤t, - Some(&previous_metadata), - &target_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, 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(¤t, &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, ¤t.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, - ¤t, - Some(&previous_metadata), - &target_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, 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(¤t_metadata, &target_metadata)?; validate_table_commit_requirements(¤t_metadata, &client_requirements)?; - validate_table_metadata_snapshot_graph( - metadata_backend, - bucket, - namespace, - &table_name, - ¤t, - Some(¤t_metadata), - &target_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, Some(¤t_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( 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, - ¤t, - 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, - ¤t, - &expected_metadata, - &request.updates, - ) - .await?; + validate_table_metadata_snapshot_graph_result(metadata_backend, bucket, ¤t, 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, ¤t, &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( + ¤t, &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(¤t_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, - ¤t, - Some(¤t_metadata), - &next_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, Some(¤t_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( + ¤t, &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, ¤t.metadata_location).await?; validate_metadata_table_location_in_bucket(bucket, ¤t_metadata)?; validate_metadata_matches_current_metadata(¤t_metadata, &target_metadata)?; - validate_table_metadata_snapshot_graph( - metadata_backend, - bucket, - namespace, - &table_name, - ¤t, - Some(¤t_metadata), - &target_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, Some(¤t_metadata), &target_metadata) + .await?; let table_bucket_fence_required = table_warehouse_location_changes(¤t, &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(¤t, &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, ¤t.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(¤t_metadata, &target_metadata)?; - validate_table_metadata_snapshot_graph( - metadata_backend, - bucket, - namespace, - &table_name, - ¤t, - None, - &target_metadata, - ) - .await?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, ¤t, None, &target_metadata).await?; let table_bucket_fence_required = table_warehouse_location_changes(¤t, &target_metadata)?; let commit_request = crate::table_catalog::TableCommitRequest { table_bucket: bucket.to_string(), diff --git a/rustfs/src/admin/handlers/table_catalog/routes.rs b/rustfs/src/admin/handlers/table_catalog/routes.rs index ceff0ca8b..fcfe781f1 100644 --- a/rustfs/src/admin/handlers/table_catalog/routes.rs +++ b/rustfs/src/admin/handlers/table_catalog/routes.rs @@ -79,6 +79,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, 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(), diff --git a/rustfs/src/admin/handlers/table_catalog/table.rs b/rustfs/src/admin/handlers/table_catalog/table.rs index fd300a2a4..21376fd81 100644 --- a/rustfs/src/admin/handlers/table_catalog/table.rs +++ b/rustfs/src/admin/handlers/table_catalog/table.rs @@ -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, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let principal = table_catalog_request_principal(&req).await?; + let request = read_bounded_json_body::( + &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] diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 3360f7d70..22ec9c087 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -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::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -940,6 +977,14 @@ fn table_catalog_ingress_requests_reject_unknown_fields() { "unexpected": true }), ); + assert_rejects_unknown_field::( + "RenameTableRequest", + serde_json::json!({ + "source": {"namespace": ["analytics"], "name": "events"}, + "destination": {"namespace": ["curated"], "name": "events_v2"}, + "unexpected": true + }), + ); assert_rejects_unknown_field::( "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.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::( + &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::( + &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(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(); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 1c14ad0e1..774ec7175 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -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", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index ba5660903..f829b8c94 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -405,6 +405,7 @@ fn expected_admin_route_matrix() -> Vec { "/{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 { "/{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( diff --git a/rustfs/src/table_catalog/error.rs b/rustfs/src/table_catalog/error.rs index 26208789c..49d26fb9f 100644 --- a/rustfs/src/table_catalog/error.rs +++ b/rustfs/src/table_catalog/error.rs @@ -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}"), diff --git a/rustfs/src/table_catalog/iceberg/commit.rs b/rustfs/src/table_catalog/iceberg/commit.rs index bfd890a6d..b8beac29a 100644 --- a/rustfs/src/table_catalog/iceberg/commit.rs +++ b/rustfs/src/table_catalog/iceberg/commit.rs @@ -269,9 +269,13 @@ pub(crate) fn record_table_commit_attempt(operation: &str) { fn table_catalog_store_result_label(result: &TableCatalogStoreResult) -> &'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", } diff --git a/rustfs/src/table_catalog/iceberg/validation.rs b/rustfs/src/table_catalog/iceberg/validation.rs index 559f3c62c..0b96f4078 100644 --- a/rustfs/src/table_catalog/iceberg/validation.rs +++ b/rustfs/src/table_catalog/iceberg/validation.rs @@ -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( 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(), diff --git a/rustfs/src/table_catalog/identifier.rs b/rustfs/src/table_catalog/identifier.rs index 279a246a8..e6c769d91 100644 --- a/rustfs/src/table_catalog/identifier.rs +++ b/rustfs/src/table_catalog/identifier.rs @@ -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 { + 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 { + 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 { + 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, diff --git a/rustfs/src/table_catalog/maintenance/recovery.rs b/rustfs/src/table_catalog/maintenance/recovery.rs index 3a7c65b2c..fc39cebec 100644 --- a/rustfs/src/table_catalog/maintenance/recovery.rs +++ b/rustfs/src/table_catalog/maintenance/recovery.rs @@ -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 { + 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, diff --git a/rustfs/src/table_catalog/mod.rs b/rustfs/src/table_catalog/mod.rs index cdcaf7d03..a909b6fad 100644 --- a/rustfs/src/table_catalog/mod.rs +++ b/rustfs/src/table_catalog/mod.rs @@ -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::*; diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index 3b7f54d1b..4e358651b 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -224,6 +224,20 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn load_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult>; + 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, diff --git a/rustfs/src/table_catalog/store/strong.rs b/rustfs/src/table_catalog/store/strong.rs index e2d525764..e1d313129 100644 --- a/rustfs/src/table_catalog/store/strong.rs +++ b/rustfs/src/table_catalog/store/strong.rs @@ -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 { 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(¤t, &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, ) -> TableCatalogStoreResult { 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); diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 1074add1a..903e33326 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -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(¤t_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(¤t_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::::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::::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(_)) + ); +}