From 604379d62fab481b73d1848bfbd045a684031441 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 18 Jun 2026 07:32:20 +0800 Subject: [PATCH] feat(table-catalog): harden table row-level commit conflicts (#3517) * feat(table-catalog): harden snapshot conflict validation * feat: harden table row-level commit conflicts * fix: accept additional row-level snapshot shapes * test(table-catalog): expand smoke catalog probes * fix(table-catalog): satisfy row-level clippy checks --------- Co-authored-by: Henry Guo Co-authored-by: houseme --- rustfs/src/admin/handlers/table_catalog.rs | 1423 ++++++++++++++++- rustfs/src/table_catalog.rs | 40 +- scripts/table-catalog/README.md | 22 +- scripts/table-catalog/pyiceberg_smoke.py | 238 ++- scripts/table-catalog/test_pyiceberg_smoke.py | 161 +- 5 files changed, 1831 insertions(+), 53 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 58fcf40db..db925992d 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -39,7 +39,7 @@ use rustfs_policy::{ }; use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::{Deserialize, Serialize, de::DeserializeOwned}; -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::time::{Duration as StdDuration, Instant}; use time::{Duration, OffsetDateTime}; use uuid::Uuid; @@ -288,7 +288,7 @@ struct PutTableRefRequest { writer: Option, } -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Deserialize)] #[serde(deny_unknown_fields)] struct DeleteTableRefRequest { #[serde(default, rename = "expected-snapshot-id")] @@ -1061,6 +1061,20 @@ async fn read_json_body(mut input: Body) -> S3Result { serde_json::from_slice(&body).map_err(|err| s3_error!(InvalidRequest, "invalid JSON: {}", err)) } +async fn read_json_body_or_default(mut input: Body) -> S3Result +where + T: Default + DeserializeOwned, +{ + let body = input + .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|err| s3_error!(InvalidRequest, "failed to read request body: {}", err))?; + if body.is_empty() { + return Ok(T::default()); + } + serde_json::from_slice(&body).map_err(|err| s3_error!(InvalidRequest, "invalid JSON: {}", err)) +} + fn warehouse_from_params(params: &Params<'_, '_>) -> S3Result { let warehouse = params.get("warehouse").unwrap_or(""); if warehouse.is_empty() { @@ -1729,7 +1743,8 @@ fn view_entry_from_create_view_request( let view_uuid = Uuid::new_v4().to_string(); let warehouse_location = request.location.unwrap_or_else(|| format!("s3://{bucket}/views/{view_id}")); validate_table_location_in_bucket(bucket, &warehouse_location)?; - let metadata_location = crate::table_catalog::default_view_metadata_file_path(namespace, &view, "00001.metadata.json"); + let metadata_location = + crate::table_catalog::default_view_metadata_file_path(namespace, &view, &next_metadata_file_name(1, &view_id)); let entry = crate::table_catalog::ViewEntry { version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, @@ -2008,6 +2023,7 @@ fn validate_table_commit_requirements(metadata: &serde_json::Value, requirements )?; } "assert-ref-snapshot-id" => validate_ref_snapshot_requirement(metadata, requirement)?, + "assert-current-snapshot-id" => validate_current_snapshot_requirement(metadata, requirement)?, _ => return Err(s3_error!(NotImplemented, "unsupported commit requirement: {requirement_type}")), } } @@ -2070,6 +2086,24 @@ fn validate_ref_snapshot_requirement(metadata: &serde_json::Value, requirement: Ok(()) } +fn validate_current_snapshot_requirement(metadata: &serde_json::Value, requirement: &serde_json::Value) -> S3Result<()> { + let actual = metadata.get("current-snapshot-id").and_then(serde_json::Value::as_i64); + if requirement.get("snapshot-id").is_some_and(serde_json::Value::is_null) { + if actual.is_some() { + return Err(s3_error!(PreconditionFailed, "commit requirement failed: current snapshot exists")); + } + return Ok(()); + } + let expected = requirement + .get("snapshot-id") + .and_then(serde_json::Value::as_i64) + .ok_or_else(|| s3_error!(InvalidRequest, "assert-current-snapshot-id requires snapshot-id"))?; + if actual != Some(expected) { + return Err(s3_error!(PreconditionFailed, "commit requirement failed: current snapshot changed")); + } + Ok(()) +} + fn apply_table_commit_updates( mut metadata: serde_json::Value, updates: &[serde_json::Value], @@ -2381,16 +2415,10 @@ fn apply_add_snapshot_update(metadata: &mut serde_json::Value, update: &serde_js .get("timestamp-ms") .and_then(serde_json::Value::as_i64) .unwrap_or_else(current_time_millis); + validate_added_snapshot(metadata, &snapshot, snapshot_id, sequence_number)?; ensure_array_field(metadata, "snapshots")?.push(snapshot); let object = metadata_object_mut(metadata)?; - let current_sequence_number = object - .get("last-sequence-number") - .and_then(serde_json::Value::as_i64) - .unwrap_or_default(); - object.insert( - "last-sequence-number".to_string(), - serde_json::Value::from(current_sequence_number.max(sequence_number)), - ); + object.insert("last-sequence-number".to_string(), serde_json::Value::from(sequence_number)); object.insert("current-snapshot-id".to_string(), serde_json::Value::from(snapshot_id)); ensure_array_field(metadata, "snapshot-log")?.push(serde_json::json!({ "timestamp-ms": timestamp_ms, @@ -2399,6 +2427,450 @@ fn apply_add_snapshot_update(metadata: &mut serde_json::Value, update: &serde_js Ok(()) } +fn validate_added_snapshot( + metadata: &serde_json::Value, + snapshot: &serde_json::Value, + snapshot_id: i64, + sequence_number: i64, +) -> S3Result<()> { + if metadata + .get("snapshots") + .and_then(serde_json::Value::as_array) + .is_some_and(|snapshots| { + snapshots + .iter() + .any(|snapshot| snapshot.get("snapshot-id").and_then(serde_json::Value::as_i64) == Some(snapshot_id)) + }) + { + return Err(s3_error!(PreconditionFailed, "snapshot id already exists")); + } + + let current_snapshot_id = metadata.get("current-snapshot-id").and_then(serde_json::Value::as_i64); + if let Some(parent_snapshot_id) = snapshot.get("parent-snapshot-id").and_then(serde_json::Value::as_i64) + && Some(parent_snapshot_id) != current_snapshot_id + { + return Err(s3_error!(PreconditionFailed, "snapshot parent no longer matches current snapshot")); + } + + let current_sequence_number = metadata + .get("last-sequence-number") + .and_then(serde_json::Value::as_i64) + .unwrap_or_default(); + if sequence_number <= current_sequence_number { + return Err(s3_error!(PreconditionFailed, "snapshot sequence number must advance")); + } + + if !snapshot_has_manifest_references(snapshot) { + return Err(s3_error!(InvalidRequest, "snapshot manifest-list or manifests are required")); + } + + let operation = snapshot + .get("summary") + .and_then(|summary| summary.get("operation")) + .and_then(serde_json::Value::as_str) + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot summary.operation is required"))?; + if !matches!(operation, "append" | "overwrite" | "delete" | "replace") { + return Err(s3_error!(NotImplemented, "unsupported snapshot operation: {operation}")); + } + + Ok(()) +} + +fn snapshot_has_manifest_references(snapshot: &serde_json::Value) -> bool { + if snapshot + .get("manifest-list") + .and_then(serde_json::Value::as_str) + .is_some_and(|manifest_list| !manifest_list.is_empty()) + { + return true; + } + snapshot + .get("manifests") + .and_then(serde_json::Value::as_array) + .is_some_and(|manifests| { + !manifests.is_empty() + && manifests + .iter() + .all(|manifest| manifest.as_str().is_some_and(|manifest| !manifest.is_empty())) + }) +} + +#[derive(Default)] +struct SnapshotLiveFiles { + data_files: BTreeSet, + delete_files: BTreeSet, +} + +impl SnapshotLiveFiles { + fn contains(&self, location: &str) -> bool { + self.data_files.contains(location) || self.delete_files.contains(location) + } +} + +#[derive(Default)] +struct SnapshotFileChanges { + added_data_files: BTreeSet, + added_delete_files: BTreeSet, + deleted_data_files: BTreeSet, + deleted_delete_files: BTreeSet, +} + +impl SnapshotFileChanges { + fn has_delete_or_row_level_change(&self) -> bool { + !self.added_delete_files.is_empty() || !self.deleted_data_files.is_empty() || !self.deleted_delete_files.is_empty() + } + + fn has_any_change(&self) -> bool { + !self.added_data_files.is_empty() || !self.added_delete_files.is_empty() || self.has_deleted_files() + } + + fn has_deleted_files(&self) -> bool { + !self.deleted_data_files.is_empty() || !self.deleted_delete_files.is_empty() + } +} + +#[derive(Clone, Copy)] +struct SnapshotChangeIdentity { + snapshot_id: i64, + sequence_number: i64, +} + +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], +) -> S3Result<()> +where + B: crate::table_catalog::TableCatalogObjectBackend, +{ + let Some(snapshot) = added_snapshot_update(updates)? else { + return Ok(()); + }; + let snapshot_id = snapshot + .get("snapshot-id") + .and_then(serde_json::Value::as_i64) + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot-id must be an integer"))?; + let sequence_number = snapshot + .get("sequence-number") + .and_then(serde_json::Value::as_i64) + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot sequence-number must be an integer"))?; + let operation = snapshot + .get("summary") + .and_then(|summary| summary.get("operation")) + .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 changes = load_snapshot_file_changes( + metadata_backend, + bucket, + namespace, + table, + entry, + snapshot, + SnapshotChangeIdentity { + snapshot_id, + sequence_number, + }, + ) + .await?; + + for location in changes.added_data_files.iter().chain(changes.added_delete_files.iter()) { + if current_live_files.contains(location) { + return Err(s3_error!( + PreconditionFailed, + "commit requirement failed: added file already exists in current snapshot" + )); + } + } + + match operation { + "append" => { + if changes.has_deleted_files() || !changes.added_delete_files.is_empty() { + return Err(s3_error!(InvalidRequest, "append snapshot cannot delete data files or add delete files")); + } + } + "overwrite" | "delete" | "replace" => { + if current_metadata + .get("current-snapshot-id") + .and_then(serde_json::Value::as_i64) + .is_none() + { + return Err(s3_error!(InvalidRequest, "row-level snapshot operation requires a current snapshot")); + } + if operation == "overwrite" { + if !changes.has_any_change() { + return Err(s3_error!(InvalidRequest, "overwrite snapshot operation requires changed files")); + } + } else if !changes.has_delete_or_row_level_change() { + return Err(s3_error!( + InvalidRequest, + "row-level snapshot operation requires deleted data files or added delete files" + )); + } + for location in changes.deleted_data_files.iter().chain(changes.deleted_delete_files.iter()) { + if !current_live_files.contains(location) { + return Err(s3_error!(PreconditionFailed, "commit requirement failed: deleted file is not current")); + } + } + } + _ => return Err(s3_error!(NotImplemented, "unsupported snapshot operation: {operation}")), + } + + Ok(()) +} + +fn added_snapshot_update(updates: &[serde_json::Value]) -> S3Result> { + let mut snapshot = None; + for update in updates { + if update.get("action").and_then(serde_json::Value::as_str) != Some("add-snapshot") { + continue; + } + if snapshot.is_some() { + return Err(s3_error!(InvalidRequest, "standard commit supports one add-snapshot update")); + } + snapshot = Some( + update + .get("snapshot") + .ok_or_else(|| s3_error!(InvalidRequest, "add-snapshot requires snapshot"))?, + ); + } + Ok(snapshot) +} + +async fn load_current_snapshot_live_files( + 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 +where + B: crate::table_catalog::TableCatalogObjectBackend, +{ + let Some(current_snapshot_id) = current_metadata + .get("current-snapshot-id") + .and_then(serde_json::Value::as_i64) + else { + return Ok(SnapshotLiveFiles::default()); + }; + let snapshot = current_metadata + .get("snapshots") + .and_then(serde_json::Value::as_array) + .and_then(|snapshots| { + snapshots + .iter() + .find(|snapshot| snapshot.get("snapshot-id").and_then(serde_json::Value::as_i64) == Some(current_snapshot_id)) + }) + .ok_or_else(|| s3_error!(InvalidRequest, "current snapshot metadata is missing"))?; + + let mut live_files = SnapshotLiveFiles::default(); + for reference in read_snapshot_manifest_references(metadata_backend, bucket, namespace, table, entry, snapshot).await? { + let status = reference + .entry_status + .ok_or_else(|| s3_error!(InvalidRequest, "manifest entry status is required"))?; + match status { + 0 | 1 => match reference.object_kind { + crate::table_catalog::TableMetadataMaintenanceObjectKind::DataFile => { + live_files.data_files.insert(reference.location); + } + crate::table_catalog::TableMetadataMaintenanceObjectKind::DeleteFile => { + live_files.delete_files.insert(reference.location); + } + _ => {} + }, + 2 => {} + _ => return Err(s3_error!(InvalidRequest, "manifest entry status is unsupported")), + } + } + Ok(live_files) +} + +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, + change_identity: SnapshotChangeIdentity, +) -> S3Result +where + B: crate::table_catalog::TableCatalogObjectBackend, +{ + let mut changes = SnapshotFileChanges::default(); + for reference in read_snapshot_manifest_references(metadata_backend, bucket, namespace, table, entry, snapshot).await? { + let status = reference + .entry_status + .ok_or_else(|| s3_error!(InvalidRequest, "manifest entry status is required"))?; + if matches!(status, 1 | 2) + && (reference.snapshot_id != Some(change_identity.snapshot_id) + || reference.sequence_number != Some(change_identity.sequence_number)) + { + return Err(s3_error!( + InvalidRequest, + "manifest changed entries must belong to the committed snapshot" + )); + } + + match (status, reference.object_kind) { + (0, _) => {} + (1, crate::table_catalog::TableMetadataMaintenanceObjectKind::DataFile) => { + changes.added_data_files.insert(reference.location); + } + (1, crate::table_catalog::TableMetadataMaintenanceObjectKind::DeleteFile) => { + changes.added_delete_files.insert(reference.location); + } + (2, crate::table_catalog::TableMetadataMaintenanceObjectKind::DataFile) => { + changes.deleted_data_files.insert(reference.location); + } + (2, crate::table_catalog::TableMetadataMaintenanceObjectKind::DeleteFile) => { + changes.deleted_delete_files.insert(reference.location); + } + _ => return Err(s3_error!(InvalidRequest, "manifest entry status is unsupported")), + } + } + Ok(changes) +} + +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 mut references = Vec::new(); + for manifest_location in manifest_locations { + let manifest_key = table_commit_object_key( + bucket, + namespace, + table, + entry, + &manifest_location, + crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestFile, + )?; + let manifest_object = metadata_backend + .read_object(bucket, &manifest_key) + .await + .map_err(catalog_store_error)? + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest object is missing"))?; + let file_references = + crate::table_catalog::data_file_references_from_manifest_avro(&manifest_object.data).map_err(catalog_store_error)?; + for reference in file_references { + validate_manifest_data_file_reference(metadata_backend, bucket, namespace, table, entry, &reference).await?; + references.push(reference); + } + } + Ok(references) +} + +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> +where + B: crate::table_catalog::TableCatalogObjectBackend, +{ + 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, + )?; + let manifest_list_object = metadata_backend + .read_object(bucket, &manifest_list_key) + .await + .map_err(catalog_store_error)? + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest-list object is missing"))?; + let references = crate::table_catalog::manifest_list_references_from_manifest_list_avro(&manifest_list_object.data) + .map_err(catalog_store_error)?; + if references.is_empty() { + return Err(s3_error!(InvalidRequest, "snapshot manifest-list must reference at least one manifest")); + } + return Ok(references.into_iter().map(|reference| reference.manifest_path).collect()); + } + + let Some(manifests) = snapshot.get("manifests").and_then(serde_json::Value::as_array) else { + return Err(s3_error!(InvalidRequest, "snapshot manifest-list is required")); + }; + if manifests.is_empty() { + return Err(s3_error!(InvalidRequest, "snapshot manifests must reference at least one manifest")); + } + manifests + .iter() + .map(|manifest| { + manifest + .as_str() + .filter(|manifest| !manifest.is_empty()) + .map(str::to_string) + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest location must be a string")) + }) + .collect() +} + +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())?; + 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 + .object_exists(bucket, &object_key) + .await + .map_err(catalog_store_error)? + { + return Err(s3_error!(InvalidRequest, "manifest referenced data file is missing")); + } + Ok(()) +} + +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, +) -> S3Result { + let object_key = crate::table_catalog::table_catalog_object_key_from_location(bucket, location) + .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) + .ok_or_else(|| s3_error!(InvalidRequest, "snapshot object is outside the table warehouse"))?; + if object_kind != expected_kind { + return Err(s3_error!(InvalidRequest, "snapshot object kind does not match manifest metadata")); + } + Ok(object_key) +} + fn apply_set_snapshot_ref_update(metadata: &mut serde_json::Value, update: &serde_json::Value) -> S3Result<()> { let ref_name = update .get("ref-name") @@ -3068,14 +3540,24 @@ where else { return Err(s3_error!(InvalidRequest, "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 current_metadata = read_table_metadata_json(metadata_backend, bucket, ¤t.metadata_location).await?; validate_table_commit_requirements(¤t_metadata, &request.requirements)?; let expected_metadata = current_metadata.clone(); let next_metadata = apply_table_commit_updates(current_metadata, &request.updates, ¤t.metadata_location)?; validate_metadata_table_location_in_bucket(bucket, &next_metadata)?; validate_metadata_matches_current_metadata(&expected_metadata, &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_snapshot_commit_conflicts( + metadata_backend, + bucket, + namespace, + &table_name, + ¤t, + &expected_metadata, + &request.updates, + ) + .await?; let (commit_id, metadata_file_token) = standard_commit_ids(request.commit_id); let next_generation = current.generation.saturating_add(1); let next_metadata_location = crate::table_catalog::default_table_metadata_file_path( @@ -4024,7 +4506,7 @@ impl Operation for DeleteTableRefHandler { let ref_name = ref_name_from_params(¶ms)?; let resource = TableCatalogResource::table(&warehouse, &namespace, &table); authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?; - let request = read_json_body::(req.input).await?; + let request = read_json_body_or_default::(req.input).await?; let metadata_backend = table_catalog_backend()?; let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); let response = @@ -5168,6 +5650,12 @@ mod tests { .as_str() .expect("created metadata should have table uuid") .to_string(); + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await; let commit_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ "requirements": [ @@ -5193,7 +5681,7 @@ mod tests { "snapshot-id": 10, "sequence-number": 1, "timestamp-ms": 1234, - "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro", + "manifest-list": manifest_list, "summary": { "operation": "append" } @@ -6130,6 +6618,712 @@ mod tests { assert!(validate_table_commit_requirements(&metadata, &requirements).is_err()); } + #[test] + fn snapshot_conflict_requirements_validate_current_snapshot_id() { + let metadata = serde_json::json!({ + "current-snapshot-id": 10 + }); + + let matching = vec![serde_json::json!({ + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + })]; + validate_table_commit_requirements(&metadata, &matching).expect("matching current snapshot should pass"); + + let stale = vec![serde_json::json!({ + "type": "assert-current-snapshot-id", + "snapshot-id": 9 + })]; + assert!(validate_table_commit_requirements(&metadata, &stale).is_err()); + + let no_snapshot_metadata = serde_json::json!({}); + let create_like = vec![serde_json::json!({ + "type": "assert-current-snapshot-id", + "snapshot-id": null + })]; + validate_table_commit_requirements(&no_snapshot_metadata, &create_like) + .expect("null current snapshot requirement should pass when no current snapshot exists"); + } + + #[test] + fn snapshot_conflict_rejects_stale_parent_or_sequence_number() { + let metadata = serde_json::json!({ + "current-snapshot-id": 10, + "last-sequence-number": 4, + "snapshots": [ + { + "snapshot-id": 10, + "sequence-number": 4, + "timestamp-ms": 1234, + "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro", + "summary": { + "operation": "append" + } + } + ], + "snapshot-log": [], + "metadata-log": [] + }); + + let stale_parent = vec![serde_json::json!({ + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 9, + "sequence-number": 5, + "timestamp-ms": 2234, + "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-11.avro", + "summary": { + "operation": "append" + } + } + })]; + assert!(apply_table_commit_updates(metadata.clone(), &stale_parent, "metadata/00001.metadata.json").is_err()); + + let stale_sequence = vec![serde_json::json!({ + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 4, + "timestamp-ms": 2234, + "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-11.avro", + "summary": { + "operation": "append" + } + } + })]; + assert!(apply_table_commit_updates(metadata, &stale_sequence, "metadata/00001.metadata.json").is_err()); + } + + #[test] + fn snapshot_conflict_rejects_unknown_snapshot_operations() { + let metadata = serde_json::json!({ + "current-snapshot-id": 10, + "last-sequence-number": 4, + "snapshots": [ + { + "snapshot-id": 10, + "sequence-number": 4, + "timestamp-ms": 1234, + "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro", + "summary": { + "operation": "append" + } + } + ], + "snapshot-log": [], + "metadata-log": [] + }); + + let updates = vec![serde_json::json!({ + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 5, + "timestamp-ms": 2234, + "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-11.avro", + "summary": { + "operation": "unknown" + } + } + })]; + assert!(apply_table_commit_updates(metadata, &updates, "metadata/00001.metadata.json").is_err()); + } + + #[tokio::test] + async fn row_level_conflict_allows_overwrite_when_deleted_file_is_current() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current_manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let old_data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + ¤t_manifest_list, + 10, + 1, + &[(&old_data_file, 0, 1, 10, 1)], + ) + .await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": current_manifest_list, + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("append commit should succeed"); + + let overwrite_manifest_list = format!("{table_location}/metadata/snap-11.avro"); + let replacement_data_file = format!("{table_location}/data/part-11.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + &overwrite_manifest_list, + 11, + 2, + &[(&old_data_file, 0, 2, 11, 2), (&replacement_data_file, 0, 1, 11, 2)], + ) + .await; + let overwrite_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 2, + "timestamp-ms": 2234, + "manifest-list": overwrite_manifest_list, + "summary": { + "operation": "overwrite" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 11, + "type": "branch" + } + ] + })) + .expect("overwrite request should parse"); + + let commit = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", overwrite_request) + .await + .expect("overwrite commit should pass manifest conflict validation"); + + assert_eq!(commit.metadata["current-snapshot-id"], 11); + assert_eq!(commit.metadata["last-sequence-number"], 2); + } + + #[tokio::test] + async fn row_level_conflict_allows_v1_manifest_snapshot() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let manifest = format!("{table_location}/metadata/manifest-10.avro"); + let data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_manifest(&metadata_backend, "warehouse", &manifest, &[(&data_file, 0, 1, 10, 1)]).await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifests": [ + manifest + ], + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + + let commit = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("v1 manifests snapshot should commit"); + + assert_eq!(commit.metadata["current-snapshot-id"], 10); + assert_eq!(commit.metadata["last-sequence-number"], 1); + } + + #[tokio::test] + async fn row_level_conflict_allows_add_only_overwrite_snapshot() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current_manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let current_data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + ¤t_manifest_list, + 10, + 1, + &[(¤t_data_file, 0, 1, 10, 1)], + ) + .await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": current_manifest_list, + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("append commit should succeed"); + + let overwrite_manifest_list = format!("{table_location}/metadata/snap-11.avro"); + let added_data_file = format!("{table_location}/data/part-11.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + &overwrite_manifest_list, + 11, + 2, + &[(&added_data_file, 0, 1, 11, 2)], + ) + .await; + let overwrite_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 2, + "timestamp-ms": 2234, + "manifest-list": overwrite_manifest_list, + "summary": { + "operation": "overwrite" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 11, + "type": "branch" + } + ] + })) + .expect("overwrite request should parse"); + + let commit = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", overwrite_request) + .await + .expect("add-only overwrite should pass conflict validation"); + + assert_eq!(commit.metadata["current-snapshot-id"], 11); + assert_eq!(commit.metadata["last-sequence-number"], 2); + } + + #[tokio::test] + async fn row_level_conflict_rejects_delete_of_non_current_file() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current_manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let current_data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + ¤t_manifest_list, + 10, + 1, + &[(¤t_data_file, 0, 1, 10, 1)], + ) + .await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": current_manifest_list, + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("append commit should succeed"); + let committed = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + + let stale_data_file = format!("{table_location}/data/stale.parquet"); + let stale_key = test_snapshot_object_key("warehouse", &stale_data_file); + metadata_backend.put_bytes("warehouse", &stale_key, b"stale".to_vec()).await; + let overwrite_manifest_list = format!("{table_location}/metadata/snap-11.avro"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + &overwrite_manifest_list, + 11, + 2, + &[(&stale_data_file, 0, 2, 11, 2)], + ) + .await; + let overwrite_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 2, + "timestamp-ms": 2234, + "manifest-list": overwrite_manifest_list, + "summary": { + "operation": "overwrite" + } + } + } + ] + })) + .expect("overwrite request should parse"); + + let error = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", overwrite_request) + .await + .expect_err("stale row-level delete should conflict"); + + assert_eq!(error.code(), &s3s::S3ErrorCode::PreconditionFailed); + let unchanged = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should still exist"); + assert_eq!(unchanged.metadata_location, committed.metadata_location); + assert_eq!(unchanged.version_token, committed.version_token); + assert_eq!(unchanged.generation, committed.generation); + } + + #[tokio::test] + async fn row_level_conflict_rejects_append_with_delete_files() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + let manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let delete_file = format!("{table_location}/delete/delete-10.parquet"); + seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&delete_file, 1, 1, 10, 1)]).await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": manifest_list, + "summary": { + "operation": "append" + } + } + } + ] + })) + .expect("append request should parse"); + + let error = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect_err("append must not add delete files"); + + assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidRequest); + let unchanged = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should still exist"); + assert_eq!(unchanged.metadata_location, current.metadata_location); + assert_eq!(unchanged.version_token, current.version_token); + assert_eq!(unchanged.generation, current.generation); + } + + #[tokio::test] + async fn row_level_conflict_rejects_missing_manifest_before_pointer_update() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current_manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let current_data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + ¤t_manifest_list, + 10, + 1, + &[(¤t_data_file, 0, 1, 10, 1)], + ) + .await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": current_manifest_list, + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("append commit should succeed"); + let committed = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + let missing_manifest_list = format!("{table_location}/metadata/missing-snap-11.avro"); + let overwrite_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 2, + "timestamp-ms": 2234, + "manifest-list": missing_manifest_list, + "summary": { + "operation": "overwrite" + } + } + } + ] + })) + .expect("overwrite request should parse"); + + let error = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", overwrite_request) + .await + .expect_err("missing manifest-list should fail before pointer update"); + + assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidRequest); + let unchanged = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should still exist"); + assert_eq!(unchanged.metadata_location, committed.metadata_location); + assert_eq!(unchanged.version_token, committed.version_token); + assert_eq!(unchanged.generation, committed.generation); + } + + #[tokio::test] + async fn row_level_conflict_rejects_manifest_outside_table_warehouse() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let current_manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let current_data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest( + &metadata_backend, + "warehouse", + ¤t_manifest_list, + 10, + 1, + &[(¤t_data_file, 0, 1, 10, 1)], + ) + .await; + let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": current_manifest_list, + "summary": { + "operation": "append" + } + } + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + } + ] + })) + .expect("append request should parse"); + commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", append_request) + .await + .expect("append commit should succeed"); + let committed = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + let outside_manifest_list = "s3://warehouse/tables/other-table/metadata/snap-11.avro"; + let overwrite_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 10 + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 11, + "parent-snapshot-id": 10, + "sequence-number": 2, + "timestamp-ms": 2234, + "manifest-list": outside_manifest_list, + "summary": { + "operation": "overwrite" + } + } + } + ] + })) + .expect("overwrite request should parse"); + + let error = commit_table_response(&store, &metadata_backend, "warehouse", &namespace, "events", overwrite_request) + .await + .expect_err("outside manifest-list should fail before pointer update"); + + assert_eq!(error.code(), &s3s::S3ErrorCode::InvalidRequest); + let unchanged = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should still exist"); + assert_eq!(unchanged.metadata_location, committed.metadata_location); + assert_eq!(unchanged.version_token, committed.version_token); + assert_eq!(unchanged.generation, committed.generation); + } + + #[tokio::test] + async fn bodyless_ref_delete_uses_default_request_options() { + let request: DeleteTableRefRequest = read_json_body_or_default(Body::empty()) + .await + .expect("bodyless ref delete should use default request options"); + + assert!(request.expected_snapshot_id.is_none()); + assert!(!request.force); + assert!(request.commit_id.is_none()); + assert!(request.idempotency_key.is_none()); + assert!(request.writer.is_none()); + } + #[test] fn table_updates_reject_unknown_actions() { let metadata = serde_json::json!({ @@ -6327,6 +7521,43 @@ mod tests { .await .expect("views should list after drop"); assert!(listed.identifiers.is_empty()); + + let recreate_request: CreateViewRequest = serde_json::from_value(serde_json::json!({ + "name": "recent_events", + "schema": { + "type": "struct", + "schema-id": 0, + "fields": [ + { + "id": 1, + "name": "id", + "required": true, + "type": "long" + } + ] + }, + "view-version": { + "version-id": 1, + "schema-id": 0, + "summary": { + "engine-name": "spark" + }, + "default-catalog": "warehouse", + "default-namespace": ["analytics"], + "representations": [ + { + "type": "sql", + "sql": "SELECT id FROM analytics.events WHERE id > 100", + "dialect": "spark" + } + ] + } + })) + .expect("standard recreate view request should parse"); + let recreated = create_view_response(&store, &metadata_backend, "warehouse", &namespace, recreate_request, true) + .await + .expect("dropped view name should be reusable"); + assert_ne!(recreated.metadata_location, created.metadata_location); } #[tokio::test] @@ -6334,7 +7565,13 @@ mod tests { let store = TestTableCatalogStore::default(); let metadata_backend = TestTableCatalogObjectBackend::default(); let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); - create_standard_events_table(&store, &metadata_backend, &namespace).await; + let created = create_standard_events_table(&store, &metadata_backend, &namespace).await; + let table_location = created.metadata["location"] + .as_str() + .expect("created metadata should have table location"); + let manifest_list = format!("{table_location}/metadata/snap-10.avro"); + let data_file = format!("{table_location}/data/part-10.parquet"); + seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await; let append_request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ "updates": [ @@ -6344,7 +7581,7 @@ mod tests { "snapshot-id": 10, "sequence-number": 1, "timestamp-ms": 1234, - "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro", + "manifest-list": manifest_list, "summary": { "operation": "append" } @@ -6756,6 +7993,17 @@ mod tests { } impl TestTableCatalogObjectBackend { + async fn put_bytes(&self, bucket: &str, object: &str, data: Vec) { + self.objects.lock().await.insert( + (bucket.to_string(), object.to_string()), + crate::table_catalog::TableCatalogObject { + data, + etag: Some("etag".to_string()), + mod_time: None, + }, + ); + } + async fn put_json(&self, bucket: &str, object: &str, value: serde_json::Value) { self.put_json_with_mod_time(bucket, object, value, None).await; } @@ -6779,6 +8027,147 @@ mod tests { } } + fn test_snapshot_object_key(bucket: &str, location: &str) -> String { + crate::table_catalog::table_catalog_object_key_from_location(bucket, location) + .expect("test snapshot object location should be valid") + } + + fn test_manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec { + let schema = apache_avro::Schema::parse_str( + r#" + { + "type": "record", + "name": "manifest_file", + "fields": [ + {"name": "manifest_path", "type": "string"}, + {"name": "sequence_number", "type": "long"}, + {"name": "added_snapshot_id", "type": "long"} + ] + } + "#, + ) + .expect("manifest list avro schema should parse"); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + for manifest_path in manifest_paths { + writer + .append(apache_avro::types::Value::Record(vec![ + ( + "manifest_path".to_string(), + apache_avro::types::Value::String((*manifest_path).to_string()), + ), + ("sequence_number".to_string(), apache_avro::types::Value::Long(sequence_number)), + ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(snapshot_id)), + ])) + .expect("manifest list record should append"); + } + writer.into_inner().expect("manifest list avro bytes should flush") + } + + fn test_manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec { + let schema = apache_avro::Schema::parse_str( + r#" + { + "type": "record", + "name": "manifest_entry", + "fields": [ + {"name": "status", "type": "int"}, + {"name": "snapshot_id", "type": "long"}, + {"name": "sequence_number", "type": "long"}, + {"name": "file_sequence_number", "type": "long"}, + { + "name": "data_file", + "type": { + "type": "record", + "name": "data_file", + "fields": [ + {"name": "content", "type": "int"}, + {"name": "file_path", "type": "string"}, + {"name": "record_count", "type": "long"}, + {"name": "file_size_in_bytes", "type": "long"} + ] + } + } + ] + } + "#, + ) + .expect("manifest avro schema should parse"); + let mut writer = apache_avro::Writer::new(&schema, Vec::new()); + for (file_path, content, status, snapshot_id, sequence_number) in files { + writer + .append(apache_avro::types::Value::Record(vec![ + ("status".to_string(), apache_avro::types::Value::Int(*status)), + ("snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)), + ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)), + ("file_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)), + ( + "data_file".to_string(), + apache_avro::types::Value::Record(vec![ + ("content".to_string(), apache_avro::types::Value::Int(*content)), + ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())), + ("record_count".to_string(), apache_avro::types::Value::Long(1)), + ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)), + ]), + ), + ])) + .expect("manifest record should append"); + } + writer.into_inner().expect("manifest avro bytes should flush") + } + + async fn seed_test_snapshot_manifest( + backend: &TestTableCatalogObjectBackend, + bucket: &str, + manifest_list_location: &str, + snapshot_id: i64, + sequence_number: i64, + files: &[(&str, i32, i32, i64, i64)], + ) { + let manifest_location = manifest_list_location + .rsplit_once('/') + .map(|(prefix, name)| format!("{prefix}/manifest-{name}")) + .expect("manifest list location should include a file name"); + let manifest_key = test_snapshot_object_key(bucket, &manifest_location); + let manifest_list_key = test_snapshot_object_key(bucket, manifest_list_location); + backend + .put_bytes( + bucket, + &manifest_list_key, + test_manifest_list_avro_bytes(&[&manifest_location], sequence_number, snapshot_id), + ) + .await; + backend + .put_bytes(bucket, &manifest_key, test_manifest_avro_bytes(files)) + .await; + seed_test_manifest_data_files(backend, bucket, files).await; + } + + async fn seed_test_manifest( + backend: &TestTableCatalogObjectBackend, + bucket: &str, + manifest_location: &str, + files: &[(&str, i32, i32, i64, i64)], + ) { + let manifest_key = test_snapshot_object_key(bucket, manifest_location); + backend + .put_bytes(bucket, &manifest_key, test_manifest_avro_bytes(files)) + .await; + seed_test_manifest_data_files(backend, bucket, files).await; + } + + async fn seed_test_manifest_data_files( + backend: &TestTableCatalogObjectBackend, + bucket: &str, + files: &[(&str, i32, i32, i64, i64)], + ) { + for (file_path, _, status, _, _) in files { + if *status != 2 { + let object_key = test_snapshot_object_key(bucket, file_path); + backend.put_bytes(bucket, &object_key, b"data".to_vec()).await; + } + } + } + async fn create_standard_events_table( store: &TestTableCatalogStore, metadata_backend: &TestTableCatalogObjectBackend, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 9f0c83275..4ea1d1e00 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -921,7 +921,7 @@ fn table_warehouse_object_prefix_from_location(table_bucket: &str, warehouse_loc normalize_warehouse_object_prefix(object_prefix) } -fn table_warehouse_object_prefix(entry: &TableEntry) -> TableCatalogStoreResult { +pub(crate) fn table_warehouse_object_prefix(entry: &TableEntry) -> TableCatalogStoreResult { table_warehouse_object_prefix_from_location(&entry.table_bucket, &entry.warehouse_location) } @@ -4035,7 +4035,9 @@ fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResul .collect()) } -fn manifest_list_references_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult> { +pub(crate) fn manifest_list_references_from_manifest_list_avro( + data: &[u8], +) -> TableCatalogStoreResult> { let reader = apache_avro::Reader::new(data) .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list Avro: {err}")))?; let mut manifest_paths = Vec::new(); @@ -4061,7 +4063,7 @@ fn file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult TableCatalogStoreResult> { +pub(crate) fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult> { let reader = apache_avro::Reader::new(data) .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest Avro: {err}")))?; let mut files = Vec::new(); @@ -4136,7 +4138,7 @@ fn avro_i64_value(value: &apache_avro::types::Value) -> Option { } } -fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) -> Option { +pub(crate) fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) -> Option { let object = if let Some(location) = location.strip_prefix("s3://") { let (bucket, object) = location.split_once('/')?; if bucket != table_bucket { @@ -4159,7 +4161,7 @@ fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) -> Some(object.to_string()) } -fn table_maintenance_object_kind( +pub(crate) fn table_maintenance_object_kind( namespace: &Namespace, table: &IdentifierSegment, warehouse_object_prefix: Option<&str>, @@ -4707,21 +4709,23 @@ struct CompactedDataFile { file_sequence_number: i64, } -struct ManifestDataFileReference { - location: String, - object_kind: TableMetadataMaintenanceObjectKind, - entry_status: Option, - snapshot_id: Option, - sequence_number: Option, - file_sequence_number: Option, - record_count: Option, - file_size_bytes: Option, +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ManifestDataFileReference { + pub location: String, + pub object_kind: TableMetadataMaintenanceObjectKind, + pub entry_status: Option, + pub snapshot_id: Option, + pub sequence_number: Option, + pub file_sequence_number: Option, + pub record_count: Option, + pub file_size_bytes: Option, } -struct ManifestListReference { - manifest_path: String, - sequence_number: Option, - added_snapshot_id: Option, +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ManifestListReference { + pub manifest_path: String, + pub sequence_number: Option, + pub added_snapshot_id: Option, } struct CompactionManifestListSummary<'a> { diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index e95faf2cc..1b20edc5c 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -32,6 +32,9 @@ The smoke test covers: - create namespace and table - append two rows through PyIceberg - reload and scan the table +- probe direct REST catalog endpoints for metadata-location, table refs, + Iceberg views, maintenance config, metadata maintenance, worker run, and + catalog diagnostics - optionally drop the table and namespace The default profile uses the canonical RustFS catalog URI: @@ -104,6 +107,15 @@ temporary credentials: - `PutObject` and `GetObject` to the same bucket outside that prefix must be rejected. +The direct REST catalog probes run by default after the PyIceberg append and +scan. For deployments that intentionally expose only the core Iceberg REST +Catalog table path, skip those probes explicitly: + +```bash +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --skip-catalog-api-probes +``` + ## Machine-Readable Inventories The script can print the current conformance inventories without importing @@ -120,20 +132,23 @@ work items. They are intentionally conservative: only PyIceberg is automated by this script today; other engines are documented until a repeatable harness is added. -RustFS also exposes catalog-backed advanced Iceberg surfaces that are not part -of the PyIceberg append smoke path yet: +The smoke test also probes catalog-backed advanced Iceberg surfaces: - table refs can be listed, created or replaced, and deleted through catalog commits; refs with explicit retention policy require a forced delete, and `main` cannot be deleted - Iceberg views support basic create, list, load, replace, existence check, and drop routes with persisted view metadata and view-scoped authorization +- metadata maintenance supports safe dry-run planning and controlled worker + execution checks +- catalog diagnostics exposes the table recovery and consistency state used by + operators ## Client Matrix | Client | Current status | Claim | |---|---|---| -| PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe | +| PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe | | Spark Iceberg REST catalog | Manual-ready | create/load/append/reload should be verified against a running RustFS endpoint | | Trino Iceberg REST catalog | Documented, not automated | no write compatibility claim yet | | DuckDB Iceberg | Documented, not automated | read-path reference only | @@ -163,6 +178,7 @@ current unsupported inventory is: - snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports - automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint; built-in periodic scheduling is not claimed - compaction rewrite: controlled run-once support for unpartitioned Parquet binpack through metadata maintenance; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed +- row-level delete/update/merge commits: standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item - external catalog bridges: metadata import/register is supported, but Polaris/Glue/DLF/Hive synchronization is unsupported - multi-table transactions: not a short-term production claim diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index f8e043d7b..c835fd1ab 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -24,6 +24,7 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = ( "s3.secret-access-key", "s3.session-token", ) +TABLE_MAINTENANCE_CONFIG_VERSION = 1 PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "rustfs": { @@ -88,7 +89,7 @@ CLIENT_MATRIX: list[dict[str, str]] = [ { "client": "PyIceberg", "status": "automated", - "coverage": "create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe", + "coverage": "create namespace, create table, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe", "entrypoint": "scripts/table-catalog/pyiceberg_smoke.py", }, { @@ -156,6 +157,12 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [ "roadmap_area": "snapshot-maintenance", "expected_behavior": "metadata maintenance can plan binpack candidates and commit a safe unpartitioned Parquet rewrite through the catalog; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed", }, + { + "capability": "row-level-delete-update-merge", + "status": "catalog-conflict-validation-supported", + "roadmap_area": "row-level-conflict-detection", + "expected_behavior": "standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item", + }, { "capability": "external-catalog-bridge", "status": "metadata-import-only", @@ -189,6 +196,15 @@ class StorageCredential: config: dict[str, str] +class RestRequestError(RuntimeError): + def __init__(self, method: str, path: str, status_code: int, response_body: str) -> None: + super().__init__(f"{method} {path} failed with HTTP {status_code}: {response_body}") + self.method = method + self.path = path + self.status_code = status_code + self.response_body = response_body + + def env_or_none(key: str) -> str | None: value = os.getenv(key) if value is None or value == "": @@ -252,6 +268,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20"))) parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.") parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.") + parser.add_argument( + "--skip-catalog-api-probes", + action="store_true", + help="Skip direct REST probes for metadata-location, refs, views, maintenance, and diagnostics endpoints.", + ) parser.add_argument("--insecure", action="store_true", help="Disable TLS verification for HTTPS endpoints.") parser.add_argument("--print-client-matrix", action="store_true", help="Print the current client conformance matrix as JSON and exit.") parser.add_argument("--print-vendor-profiles", action="store_true", help="Print vendor connection profile references as JSON and exit.") @@ -362,7 +383,25 @@ def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str return json.loads(response_data.decode("utf-8")) except urllib.error.HTTPError as error: response_body = error.read().decode("utf-8", errors="replace") - raise RuntimeError(f"{method} {path} failed with HTTP {error.code}: {response_body}") from error + raise RestRequestError(method, path, error.code, response_body) from error + + +def signed_rest_request_expect_error( + args: argparse.Namespace, + deps: RuntimeDeps, + method: str, + path: str, + body: dict[str, Any] | None = None, + expected_statuses: set[int] | None = None, +) -> RestRequestError: + try: + signed_rest_request(args, deps, method, path, body) + except RestRequestError as error: + if expected_statuses is not None and error.status_code not in expected_statuses: + expected = ", ".join(str(status) for status in sorted(expected_statuses)) + raise RuntimeError(f"{method} {path} failed with HTTP {error.status_code}, expected one of: {expected}") from error + return error + raise RuntimeError(f"{method} {path} unexpectedly succeeded") def ensure_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None: @@ -406,6 +445,45 @@ def credentials_endpoint_path(args: argparse.Namespace) -> str: return f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/tables/{encoded_table}/credentials" +def table_endpoint_path(args: argparse.Namespace, suffix: str = "") -> str: + encoded_bucket = urllib.parse.quote(args.bucket, safe="") + encoded_namespace = urllib.parse.quote(args.namespace, safe="") + encoded_table = urllib.parse.quote(args.table, safe="") + return f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/tables/{encoded_table}{suffix}" + + +def table_ref_endpoint_path(args: argparse.Namespace, ref_name: str | None = None) -> str: + path = table_endpoint_path(args, "/refs") + if ref_name is None: + return path + return f"{path}/{urllib.parse.quote(ref_name, safe='')}" + + +def view_endpoint_path(args: argparse.Namespace, view_name: str | None = None) -> str: + encoded_bucket = urllib.parse.quote(args.bucket, safe="") + encoded_namespace = urllib.parse.quote(args.namespace, safe="") + path = f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/views" + if view_name is None: + return path + return f"{path}/{urllib.parse.quote(view_name, safe='')}" + + +def default_maintenance_config() -> dict[str, Any]: + return { + "version": TABLE_MAINTENANCE_CONFIG_VERSION, + "retain-recent-metadata-files": 2, + "delete-enabled": False, + "background-enabled": False, + "worker-paused": True, + "worker-lease-timeout-seconds": 60, + "max-retry-attempts": 0, + "retry-initial-backoff-seconds": 5, + "retry-max-backoff-seconds": 300, + "quarantine-enabled": False, + "quarantine-retention-seconds": 0, + } + + def storage_credential_from_response(response: dict[str, Any]) -> StorageCredential: credentials = response.get("storage-credentials") if not isinstance(credentials, list) or not credentials: @@ -692,29 +770,155 @@ def cleanup_catalog(catalog: Any, identifier: tuple[str, str]) -> None: print(f"warning: failed to drop namespace {identifier[0]}: {error}", file=sys.stderr) +def current_snapshot_id_from_table_response(response: dict[str, Any]) -> int: + metadata = response.get("metadata") + if not isinstance(metadata, dict): + raise RuntimeError("REST loadTable response did not include metadata") + snapshot_id = metadata.get("current-snapshot-id") + if not isinstance(snapshot_id, int): + raise RuntimeError("REST loadTable response did not include a current snapshot id") + return snapshot_id + + +def run_metadata_location_probe(args: argparse.Namespace, deps: RuntimeDeps, table_response: dict[str, Any]) -> None: + metadata_location = table_response.get("metadata-location") + if not isinstance(metadata_location, str) or not metadata_location: + raise RuntimeError("REST loadTable response did not include metadata-location") + location_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/metadata-location")) + if location_response.get("metadata-location") != metadata_location: + raise RuntimeError("metadata-location endpoint does not match loadTable metadata-location") + if not location_response.get("version-token"): + raise RuntimeError("metadata-location endpoint did not include version-token") + + +def run_table_ref_probe(args: argparse.Namespace, deps: RuntimeDeps, snapshot_id: int) -> None: + ref_name = f"smoke-{safe_probe_segment(args.namespace, args.table)}" + ref_body = { + "snapshot-id": snapshot_id, + "type": "tag", + "max-ref-age-ms": 86_400_000, + } + signed_rest_request(args, deps, "PUT", table_ref_endpoint_path(args, ref_name), ref_body) + refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) + if refs.get("refs", {}).get(ref_name, {}).get("snapshot-id") != snapshot_id: + raise RuntimeError("table refs endpoint did not return the smoke tag") + signed_rest_request_expect_error(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), expected_statuses={400}) + signed_rest_request(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), {"force": True}) + refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args)) + if ref_name in refs.get("refs", {}): + raise RuntimeError("table refs endpoint still returned the deleted smoke tag") + + +def smoke_view_request(args: argparse.Namespace, view_name: str, version_id: int, sql: str) -> dict[str, Any]: + return { + "name": view_name, + "schema": { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "id", "required": True, "type": "long"}, + {"id": 2, "name": "payload", "required": False, "type": "string"}, + ], + }, + "view-version": { + "version-id": version_id, + "schema-id": 0, + "timestamp-ms": int(time.time() * 1000), + "summary": {"operation": "replace"}, + "representations": [ + { + "type": "sql", + "sql": sql, + "dialect": "spark", + } + ], + }, + "properties": {"rustfs.smoke": "true", "rustfs.smoke.table": args.table}, + } + + +def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: + view_name = f"{args.table}_smoke_view" + create_body = smoke_view_request(args, view_name, 1, f"SELECT id, payload FROM {args.namespace}.{args.table}") + signed_rest_request(args, deps, "POST", view_endpoint_path(args), create_body) + try: + views = signed_rest_request(args, deps, "GET", view_endpoint_path(args)) + identifiers = views.get("identifiers", []) + if not any(identifier.get("name") == view_name for identifier in identifiers if isinstance(identifier, dict)): + raise RuntimeError("listViews response did not include the smoke view") + loaded = signed_rest_request(args, deps, "GET", view_endpoint_path(args, view_name)) + metadata_location = loaded.get("metadata-location") + if not isinstance(metadata_location, str) or not metadata_location: + raise RuntimeError("loadView response did not include metadata-location") + finally: + try: + signed_rest_request(args, deps, "DELETE", view_endpoint_path(args, view_name)) + except RestRequestError as error: + if error.status_code != 404: + raise + + +def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: + config_path = table_endpoint_path(args, "/maintenance/config") + signed_rest_request(args, deps, "PUT", config_path, default_maintenance_config()) + config = signed_rest_request(args, deps, "GET", config_path) + if config.get("version") != TABLE_MAINTENANCE_CONFIG_VERSION: + raise RuntimeError("maintenance config endpoint did not persist the smoke config") + + report = signed_rest_request( + args, + deps, + "POST", + table_endpoint_path(args, "/maintenance/metadata"), + {"retain-recent-metadata-files": 1, "delete": False}, + ) + job = report.get("job") + if not isinstance(job, dict) or not job.get("job-id"): + raise RuntimeError("maintenance metadata endpoint did not return a job id") + signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}")) + + worker = signed_rest_request(args, deps, "POST", table_endpoint_path(args, "/maintenance/worker/run"), {}) + worker_job = worker.get("job") + expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING"} + if not isinstance(worker_job, dict) or worker_job.get("status") not in expected_statuses: + raise RuntimeError("maintenance worker endpoint did not return a stable job status") + + +def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None: + table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args)) + snapshot_id = current_snapshot_id_from_table_response(table_response) + run_metadata_location_probe(args, deps, table_response) + run_table_ref_probe(args, deps, snapshot_id) + run_view_probe(args, deps) + run_maintenance_probe(args, deps) + diagnostics = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/catalog/diagnostics")) + if not isinstance(diagnostics, dict) or not diagnostics: + raise RuntimeError("catalog diagnostics endpoint returned an empty response") + + def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: endpoint = normalized_endpoint(args.endpoint) ensure_local_proxy_bypass(endpoint) ensure_aws_env(args.access_key, args.secret_key, args.region) - print(f"[1/9] ensuring S3 bucket {args.bucket}") + print(f"[1/10] ensuring S3 bucket {args.bucket}") ensure_bucket(args, deps) - print(f"[2/9] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1") + print(f"[2/10] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1") enable_table_bucket(args, deps) - print(f"[3/9] loading PyIceberg REST catalog at {endpoint}{args.rest_path}") + print(f"[3/10] loading PyIceberg REST catalog at {endpoint}{args.rest_path}") catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args)) install_rustfs_rest_sigv4_adapter(catalog, args, deps) identifier = table_identifier(args) if args.replace: - print(f"[4/9] replacing existing table {'.'.join(identifier)} if present") + print(f"[4/10] replacing existing table {'.'.join(identifier)} if present") drop_table_if_present(catalog, identifier) else: - print(f"[4/9] table {'.'.join(identifier)} is available") + print(f"[4/10] table {'.'.join(identifier)} is available") - print(f"[5/9] creating namespace and table {'.'.join(identifier)}") + print(f"[5/10] creating namespace and table {'.'.join(identifier)}") ensure_namespace(catalog, args.namespace) arrow_schema = deps.pyarrow.schema( [ @@ -726,10 +930,10 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: storage_credential = None if args.require_vended_credentials: - print(f"[6/9] loading table-scoped storage credentials") + print(f"[6/10] loading table-scoped storage credentials") storage_credential = load_table_storage_credential(args, deps) print(f"credential scope: {storage_credential.prefix or 'not reported'}") - print(f"[7/9] verifying table-scoped data-plane access") + print(f"[7/10] verifying table-scoped data-plane access") try: expected_table_location = table_warehouse_location(created_table) except RuntimeError: @@ -738,10 +942,10 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args, storage_credential=storage_credential)) install_rustfs_rest_sigv4_adapter(catalog, args, deps) else: - print(f"[6/9] using configured S3 credentials for data-plane operations") - print(f"[7/9] skipping vended credential data-plane scope probe") + print(f"[6/10] using configured S3 credentials for data-plane operations") + print(f"[7/10] skipping vended credential data-plane scope probe") - print(f"[8/9] appending rows through PyIceberg") + print(f"[8/10] appending rows through PyIceberg") table = catalog.load_table(identifier) rows = deps.pyarrow.Table.from_pylist( [ @@ -752,12 +956,18 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None: ) table.append(rows) - print(f"[9/9] reloading and scanning table") + print(f"[9/10] reloading and scanning table") loaded = catalog.load_table(identifier) scanned = loaded.scan().to_arrow() if scanned.num_rows != 2: raise RuntimeError(f"expected 2 rows after append, got {scanned.num_rows}") + if args.skip_catalog_api_probes: + print("[10/10] skipping direct REST catalog API probes") + else: + print("[10/10] probing direct REST catalog APIs") + run_catalog_api_probes(args, deps) + if args.cleanup: print(f"cleanup: dropping table and namespace {'.'.join(identifier)}") cleanup_catalog(catalog, identifier) diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 9b31ef3f9..1412b5ed3 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -100,6 +100,165 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/credentials", ) + def test_table_catalog_endpoint_paths_encode_identifier_components(self) -> None: + args = self.parse_with_args([ + "--bucket", + "lake bucket", + "--namespace", + "sales.analytics", + "--table", + "orders table", + ]) + + self.assertEqual( + pyiceberg_smoke.table_endpoint_path(args), + "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table", + ) + self.assertEqual( + pyiceberg_smoke.table_endpoint_path(args, "/metadata-location"), + "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/metadata-location", + ) + self.assertEqual( + pyiceberg_smoke.table_ref_endpoint_path(args, "release/2026"), + "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/refs/release%2F2026", + ) + self.assertEqual( + pyiceberg_smoke.view_endpoint_path(args, "orders view"), + "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/views/orders%20view", + ) + + def test_default_maintenance_config_is_safe_for_smoke_runs(self) -> None: + config = pyiceberg_smoke.default_maintenance_config() + + self.assertEqual(config["version"], 1) + self.assertFalse(config["delete-enabled"]) + self.assertFalse(config["background-enabled"]) + self.assertTrue(config["worker-paused"]) + self.assertEqual(config["max-retry-attempts"], 0) + + def test_expected_error_helper_returns_matching_rest_error(self) -> None: + args = self.parse_with_args([]) + expected = pyiceberg_smoke.RestRequestError("DELETE", "/path", 400, "bad request") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=expected): + returned = pyiceberg_smoke.signed_rest_request_expect_error( + args, + mock.Mock(), + "DELETE", + "/path", + expected_statuses={400}, + ) + + self.assertIs(returned, expected) + + def test_expected_error_helper_rejects_wrong_status_or_success(self) -> None: + args = self.parse_with_args([]) + wrong_status = pyiceberg_smoke.RestRequestError("DELETE", "/path", 404, "missing") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=wrong_status): + with self.assertRaisesRegex(RuntimeError, "expected one of"): + pyiceberg_smoke.signed_rest_request_expect_error( + args, + mock.Mock(), + "DELETE", + "/path", + expected_statuses={400}, + ) + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", return_value={}): + with self.assertRaisesRegex(RuntimeError, "unexpectedly succeeded"): + pyiceberg_smoke.signed_rest_request_expect_error(args, mock.Mock(), "DELETE", "/path") + + def test_current_snapshot_id_is_read_from_rest_load_table_response(self) -> None: + self.assertEqual( + pyiceberg_smoke.current_snapshot_id_from_table_response({"metadata": {"current-snapshot-id": 42}}), + 42, + ) + + with self.assertRaisesRegex(RuntimeError, "metadata"): + pyiceberg_smoke.current_snapshot_id_from_table_response({}) + with self.assertRaisesRegex(RuntimeError, "current snapshot id"): + pyiceberg_smoke.current_snapshot_id_from_table_response({"metadata": {}}) + + def test_catalog_api_probe_exercises_extended_rest_surfaces(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + deps = mock.Mock() + calls: list[tuple[str, str, object]] = [] + table_path = pyiceberg_smoke.table_endpoint_path(args) + metadata_location_path = pyiceberg_smoke.table_endpoint_path(args, "/metadata-location") + refs_path = pyiceberg_smoke.table_ref_endpoint_path(args) + view_path = pyiceberg_smoke.view_endpoint_path(args) + config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") + maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") + worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") + diagnostics_path = pyiceberg_smoke.table_endpoint_path(args, "/catalog/diagnostics") + + def fake_signed_request( + _args: object, + _deps: object, + method: str, + path: str, + body: object = None, + ) -> dict[str, object]: + calls.append((method, path, body)) + if (method, path) == ("GET", table_path): + return {"metadata-location": "s3://lake/tables/id/metadata/v2.metadata.json", "metadata": {"current-snapshot-id": 7}} + if (method, path) == ("GET", metadata_location_path): + return {"metadata-location": "s3://lake/tables/id/metadata/v2.metadata.json", "version-token": "token-2"} + if method == "PUT" and path.startswith(f"{refs_path}/"): + return {} + if (method, path) == ("GET", refs_path): + if sum(1 for call in calls if call[:2] == ("GET", refs_path)) == 1: + return {"refs": {"smoke-sales-orders": {"snapshot-id": 7}}} + return {"refs": {}} + if method == "DELETE" and path.startswith(f"{refs_path}/"): + return {} + if (method, path) == ("POST", view_path): + return {} + if (method, path) == ("GET", view_path): + return {"identifiers": [{"name": "orders_smoke_view"}]} + if (method, path) == ("GET", f"{view_path}/orders_smoke_view"): + return {"metadata-location": "s3://lake/views/orders_smoke_view/metadata/v1.json"} + if (method, path) == ("DELETE", f"{view_path}/orders_smoke_view"): + return {} + if (method, path) == ("PUT", config_path): + return {} + if (method, path) == ("GET", config_path): + return {"version": 1} + if (method, path) == ("POST", maintenance_path): + return {"job": {"job-id": "job-1"}} + if (method, path) == ("GET", pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")): + return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}} + if (method, path) == ("POST", worker_path): + return {"job": {"status": "PAUSED"}} + if (method, path) == ("GET", diagnostics_path): + return {"status": "ok"} + raise AssertionError(f"unexpected REST request: {method} {path}") + + with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request): + with mock.patch.object( + pyiceberg_smoke, + "signed_rest_request_expect_error", + return_value=pyiceberg_smoke.RestRequestError("DELETE", f"{refs_path}/smoke-sales-orders", 400, "retained"), + ) as expect_error: + pyiceberg_smoke.run_catalog_api_probes(args, deps) + + expect_error.assert_called_once() + self.assertIn(("GET", metadata_location_path, None), calls) + self.assertIn(("GET", diagnostics_path, None), calls) + self.assertIn(("POST", worker_path, {}), calls) + + def test_smoke_view_request_uses_stable_iceberg_view_shape(self) -> None: + args = self.parse_with_args(["--namespace", "sales", "--table", "orders"]) + + request = pyiceberg_smoke.smoke_view_request(args, "orders_view", 7, "SELECT id FROM sales.orders") + + self.assertEqual(request["name"], "orders_view") + self.assertEqual(request["schema"]["type"], "struct") + self.assertEqual(request["view-version"]["version-id"], 7) + self.assertEqual(request["view-version"]["representations"][0]["dialect"], "spark") + self.assertEqual(request["properties"]["rustfs.smoke.table"], "orders") + def test_catalog_properties_can_use_vended_storage_credentials(self) -> None: args = self.parse_with_args([ "--access-key", @@ -313,7 +472,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): capabilities = {entry["capability"] for entry in inventory} self.assertIn("credential-vending", capabilities) - self.assertIn("iceberg-views", capabilities) + self.assertIn("row-level-delete-update-merge", capabilities) self.assertIn("background-maintenance-worker", capabilities) for entry in inventory: self.assertIn("status", entry)