From e3be247b40d81e166a957a91d91d57e9ca69041e Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Thu, 17 Sep 2026 06:38:41 +0800 Subject: [PATCH] feat(tables): support staged table creation (#7919) --- crates/ecstore/src/store/heal.rs | 12 +- crates/ecstore/src/store/init.rs | 11 +- docs/architecture/s3-tables-support-matrix.md | 6 +- .../src/admin/handlers/table_catalog/mod.rs | 327 ++++++++++- .../src/admin/handlers/table_catalog/table.rs | 4 + .../src/admin/handlers/table_catalog/tests.rs | 515 +++++++++++++++++- rustfs/src/connect/diagnostics/perf_object.rs | 7 +- rustfs/src/table_catalog/store/mod.rs | 27 + rustfs/src/table_catalog/store/object.rs | 4 + rustfs/src/table_catalog/store/strong.rs | 10 + rustfs/tests/connect_perf_object.rs | 56 ++ scripts/table-catalog/README.md | 4 +- scripts/table-catalog/duckdb_smoke.py | 17 +- scripts/table-catalog/engine_compatibility.py | 12 +- scripts/table-catalog/test_duckdb_smoke.py | 11 - .../test_engine_compatibility.py | 11 +- 16 files changed, 938 insertions(+), 96 deletions(-) diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 4c83e8395..b3708cda5 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -2776,7 +2776,16 @@ mod tests { let mut active_reader = PutObjReader::from_vec(b"active owner".to_vec()); store.pools[0] - .put_object(&bucket, active_object, &mut active_reader, &ObjectOptions::default()) + .put_object( + &bucket, + active_object, + &mut active_reader, + &ObjectOptions { + // Drain the rename tail before removing a physical shard. + write_completion: crate::object_api::WriteCompletion::TailDrained, + ..Default::default() + }, + ) .await .expect("active owner object should be written"); let active_disks = store.pools[0].disk_set[0].disks.read().await.clone(); @@ -2812,6 +2821,7 @@ mod tests { &mut duplicate_reader, &ObjectOptions { mod_time: Some(OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(mod_time)), + write_completion: crate::object_api::WriteCompletion::TailDrained, ..Default::default() }, ) diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 115121e15..218c28bb1 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -9826,14 +9826,7 @@ mod tests { .await .expect("manual task receipt path should resolve"); let target_task_set = store.pools[1].get_disks_by_key(&manual_task_receipt_path); - let original_target_task_disks = { - let mut disks = target_task_set.disks.write().await; - let original = disks.clone(); - for disk in disks.iter_mut().take(2) { - *disk = None; - } - original - }; + let offline_target_task_disks = force_set_disk_range_offline_for_test(&target_task_set, 0..2).await; let receipt_quorum_error = store .verify_and_cleanup_decommissioned_durable_ilm_record_for_test( 0, @@ -9842,7 +9835,7 @@ mod tests { ) .await .expect_err("target read quorum without receipt write quorum must retain the source"); - *target_task_set.disks.write().await = original_target_task_disks; + drop(offline_target_task_disks); let receipt_quorum_error = receipt_quorum_error.to_string(); assert!(receipt_quorum_error.contains("receipt")); assert!(receipt_quorum_error.contains(&manual_task_path)); diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 7c500340d..ed21e984a 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -39,9 +39,9 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t | Client or engine | Status | Claim | |---|---|---| | PyIceberg | Automated | Namespace and table create, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional vended credentials with an exact-prefix data-plane scope check. | -| Spark Iceberg REST catalog | Manual/live harness | Pinned package inputs, catalog properties, SQL, expected `row_count=2`, and a CI opt-in gate for create/append/refresh/count/cleanup. Live execution and commit-conflict probing remain manual unless enabled in the runner. | +| Spark Iceberg REST catalog | Manual/live harness | Pinned package inputs, catalog properties, atomic CTAS SQL through staged create, expected `row_count=2`, and a CI opt-in gate for create/refresh/count/cleanup. Live execution and commit-conflict probing remain manual unless enabled in the runner. | | Trino Iceberg REST catalog | Manual/live harness | Catalog properties and a read-only `SELECT COUNT(*)` against a PyIceberg- or Spark-created table. Write compatibility not claimed. | -| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` covers metadata-location read, single-table create/insert/update/delete/merge, schema evolution, snapshots, concurrent writers, drop, PyIceberg cross-read, and both signing profiles. Staged create, purge-on-drop, and format v3 are verified fail-closed. Two-table mode runs without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES` and vended-credential integration not claimed. | +| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` covers metadata-location read, single-table create/insert/update/delete/merge, schema evolution, snapshots, concurrent writers, drop, PyIceberg cross-read, and both signing profiles. Purge-on-drop and format v3 are verified fail-closed. Two-table mode runs without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES`, staged-create client integration, and vended-credential integration are not claimed. | | StarRocks Iceberg REST catalog | Documented, not automated | External catalog read-path reference only. | | Databend | Manual/live harness | S3 stage read probe for table data files only; Iceberg REST integration not claimed. | | Snowflake Open Catalog / Iceberg integrations | Generated harness | Operator-adapted external volume/catalog SQL template only; live interop not claimed. | @@ -62,7 +62,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t | Catalog config | Supported | `GET /v1/config` advertises defaults and only the supported OpenAPI REST paths in `endpoints`; RustFS extensions (administration, maintenance, migration, diagnostics, refs, metadata-location) are not presented as standard endpoints. | | Table bucket discovery | Supported | `PUT` / `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. | | Namespaces | Supported | Create, list, load, exists, drop on both prefixes; `pageSize`/`pageToken` pagination with context-bound tokens; identifiers limited to 512 ASCII characters. | -| Tables | Supported | Create, register, list, load, exists, rename, commit, metadata-location get/update, drop on both prefixes. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. | +| Tables | Supported | Direct and staged create, register, list, load, exists, rename, commit, metadata-location get/update, and drop on both prefixes. Staged create is stateless until an `assert-create` commit atomically publishes the initial metadata and table entry. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. | | Commit CAS | Supported | Single-table commits validate base metadata, version token, referenced object existence, warehouse scope, and Iceberg requirements before advancing the pointer; external metadata transitions preserve monotonic assignment watermarks and immutable retained definitions. `idempotency-key-lifetime` is not advertised; mutation-wide `Idempotency-Key` semantics are unsupported. | | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. | | Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. | diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 1c4127a63..54949393d 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -910,6 +910,16 @@ struct RestLoadTableResponse { storage_credentials: Vec, } +#[derive(Debug, Serialize)] +struct RestCreateTableResponse { + #[serde(rename = "metadata-location")] + metadata_location: Option, + metadata: serde_json::Value, + config: BTreeMap, + #[serde(rename = "storage-credentials")] + storage_credentials: Vec, +} + #[derive(Debug, Serialize)] struct RestLoadCredentialsResponse { config: BTreeMap, @@ -2706,6 +2716,25 @@ fn load_table_response_from_entry(entry: crate::table_catalog::TableEntry, metad } } +fn create_table_response_from_entry( + entry: crate::table_catalog::TableEntry, + metadata: serde_json::Value, + staged: bool, +) -> RestCreateTableResponse { + let RestLoadTableResponse { + metadata_location, + metadata, + config, + storage_credentials, + } = load_table_response_from_entry(entry, metadata); + RestCreateTableResponse { + metadata_location: (!staged).then_some(metadata_location), + metadata, + config, + storage_credentials, + } +} + fn load_view_response_from_entry(entry: crate::table_catalog::ViewEntry, metadata: serde_json::Value) -> RestLoadViewResponse { let mut config = BTreeMap::new(); let warehouse_location = entry.warehouse_location.clone(); @@ -3104,16 +3133,9 @@ fn table_entry_from_create_table_request( mut schema, mut partition_spec, mut write_order, - stage_create, + stage_create: _, mut properties, } = request; - if stage_create { - return Err(iceberg_rest_error( - ICEBERG_ERROR_UNSUPPORTED_OPERATION, - StatusCode::NOT_ACCEPTABLE, - "stage-create is not supported", - )); - } let table = crate::table_catalog::IdentifierSegment::parse(name) .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; @@ -3490,6 +3512,26 @@ fn validate_table_commit_requirements(metadata: &serde_json::Value, requirements Ok(()) } +fn request_has_assert_create_requirement(request: &RestCommitTableRequest) -> bool { + request + .requirements + .iter() + .any(|requirement| requirement.get("type").and_then(serde_json::Value::as_str) == Some("assert-create")) +} + +fn validate_create_table_commit_requirements(requirements: &[serde_json::Value]) -> S3Result<()> { + if requirements.is_empty() + || requirements + .iter() + .any(|requirement| requirement.get("type").and_then(serde_json::Value::as_str) != Some("assert-create")) + { + return Err(S3Error::from(ApiError::invalid_request( + "create table commit requires only assert-create requirements", + ))); + } + Ok(()) +} + fn validate_i64_requirement( metadata: &serde_json::Value, requirement: &serde_json::Value, @@ -3555,20 +3597,104 @@ fn apply_table_commit_updates( } fn apply_table_commit_updates_at( - mut metadata: serde_json::Value, + metadata: serde_json::Value, updates: &[serde_json::Value], previous_metadata_location: &str, commit_timestamp_ms: i64, +) -> S3Result { + apply_table_updates_at( + metadata, + updates, + TableMetadataUpdateMode::Existing { + previous_metadata_location, + }, + commit_timestamp_ms, + ) +} + +fn apply_table_create_updates_at(updates: &[serde_json::Value], commit_timestamp_ms: i64) -> S3Result { + let format_version = updates + .iter() + .find(|update| update.get("action").and_then(serde_json::Value::as_str) == Some("upgrade-format-version")) + .and_then(|update| update.get("format-version")) + .and_then(serde_json::Value::as_i64) + .unwrap_or(2); + let mut metadata = serde_json::json!({ + "format-version": format_version, + "last-updated-ms": commit_timestamp_ms, + "last-column-id": 0, + "schemas": [], + "current-schema-id": -1, + "partition-specs": [], + "default-spec-id": -1, + "last-partition-id": 999, + "sort-orders": [], + "default-sort-order-id": -1, + "properties": {}, + "current-snapshot-id": -1, + "snapshots": [], + "snapshot-log": [], + "metadata-log": [], + "refs": {} + }); + if format_version == 2 { + metadata_object_mut(&mut metadata)?.insert("last-sequence-number".to_string(), serde_json::Value::from(0)); + } + apply_table_updates_at(metadata, updates, TableMetadataUpdateMode::Create, commit_timestamp_ms) +} + +#[derive(Debug, Clone, Copy)] +enum TableMetadataUpdateMode<'a> { + Create, + Existing { previous_metadata_location: &'a str }, +} + +impl TableMetadataUpdateMode<'_> { + fn initial_catalog_id(self, array_key: &str) -> Option { + match (self, array_key) { + (Self::Create, "schemas" | "partition-specs") => Some(0), + (Self::Create, "sort-orders") => Some(1), + (Self::Create, _) | (Self::Existing { .. }, _) => None, + } + } +} + +fn apply_table_updates_at( + mut metadata: serde_json::Value, + updates: &[serde_json::Value], + mode: TableMetadataUpdateMode<'_>, + commit_timestamp_ms: i64, ) -> S3Result { if !metadata.is_object() { return Err(s3_error!(InvalidRequest, "current table metadata must be a JSON object")); } - if metadata.get("format-version").is_some() { + if matches!(mode, TableMetadataUpdateMode::Existing { .. }) && metadata.get("format-version").is_some() { crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?; } - let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id")?; - let mut next_spec_id = next_catalog_id_for_updates(&metadata, updates, "add-spec", "partition-specs", "spec-id")?; - let mut next_sort_order_id = next_catalog_id_for_updates(&metadata, updates, "add-sort-order", "sort-orders", "order-id")?; + let mut next_schema_id = next_catalog_id_for_updates( + &metadata, + updates, + "add-schema", + "schemas", + "schema-id", + mode.initial_catalog_id("schemas"), + )?; + let mut next_spec_id = next_catalog_id_for_updates( + &metadata, + updates, + "add-spec", + "partition-specs", + "spec-id", + mode.initial_catalog_id("partition-specs"), + )?; + let mut next_sort_order_id = next_catalog_id_for_updates( + &metadata, + updates, + "add-sort-order", + "sort-orders", + "order-id", + mode.initial_catalog_id("sort-orders"), + )?; let mut last_added_schema_id = None; let mut last_added_spec_id = None; let mut last_added_sort_order_id = None; @@ -3658,8 +3784,16 @@ fn apply_table_commit_updates_at( if metadata.get("format-version").is_some() { crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?; } - append_previous_metadata_log(&mut metadata, previous_metadata_location)?; + if let TableMetadataUpdateMode::Existing { + previous_metadata_location, + } = mode + { + append_previous_metadata_log(&mut metadata, previous_metadata_location)?; + } metadata_object_mut(&mut metadata)?.insert("last-updated-ms".to_string(), serde_json::Value::from(commit_timestamp_ms)); + if matches!(mode, TableMetadataUpdateMode::Create) { + crate::table_catalog::validate_supported_table_metadata(&metadata).map_err(catalog_store_error)?; + } Ok(metadata) } @@ -3726,7 +3860,7 @@ fn apply_view_commit_updates_at( if !metadata.is_object() { return Err(s3_error!(InvalidRequest, "current view metadata must be a JSON object")); } - let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id")?; + let mut next_schema_id = next_catalog_id_for_updates(&metadata, updates, "add-schema", "schemas", "schema-id", None)?; let mut last_added_schema_id = None; let mut last_added_view_version_id = None; let mut added_view_version_timestamps = BTreeMap::new(); @@ -5084,11 +5218,12 @@ fn next_catalog_id_for_updates( action: &str, array_key: &str, id_key: &str, + initial_id: Option, ) -> S3Result> { updates .iter() .any(|update| update.get("action").and_then(serde_json::Value::as_str) == Some(action)) - .then(|| next_array_object_i64(metadata, array_key, id_key)) + .then(|| next_array_object_i64(metadata, array_key, id_key, initial_id)) .transpose() } @@ -5103,10 +5238,22 @@ fn take_catalog_assigned_id(next_id: &mut Option, label: &str) -> S3Result< Ok(assigned_id) } -fn next_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str) -> S3Result { - last_array_object_i64(metadata, array_key, id_key)? - .checked_add(1) - .ok_or_else(|| s3_error!(InvalidRequest, "metadata field {array_key} {id_key} exceeds the signed 64-bit range")) +fn next_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str, initial_id: Option) -> S3Result { + match last_array_object_i64(metadata, array_key, id_key) { + Ok(last_id) => last_id + .checked_add(1) + .ok_or_else(|| s3_error!(InvalidRequest, "metadata field {array_key} {id_key} exceeds the signed 64-bit range")), + Err(_) + if metadata + .get(array_key) + .and_then(serde_json::Value::as_array) + .is_some_and(Vec::is_empty) => + { + initial_id + .ok_or_else(|| S3Error::from(ApiError::invalid_request(format!("metadata field {array_key} has no {id_key}")))) + } + Err(err) => Err(err), + } } fn last_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str) -> S3Result { @@ -5435,11 +5582,40 @@ async fn create_table_response( namespace: &crate::table_catalog::Namespace, request: CreateTableRequest, table_bucket_enabled: bool, -) -> S3Result +) -> S3Result where S: crate::table_catalog::TableCatalogStore + ?Sized, { + let staged = request.stage_create; let (entry, metadata) = table_entry_from_create_table_request(bucket, namespace, request)?; + if staged { + if !table_bucket_enabled { + return Err(S3Error::from(ApiError::invalid_request(format!("bucket {bucket} is not table-enabled")))); + } + get_namespace_response(store, bucket, namespace).await?; + if store + .load_table(bucket, &namespace.public_name(), &entry.table) + .await + .map_err(catalog_store_error)? + .is_some() + || store + .load_view(bucket, &namespace.public_name(), &entry.table) + .await + .map_err(catalog_store_error)? + .is_some() + { + return Err(iceberg_rest_error( + ICEBERG_ERROR_ALREADY_EXISTS, + StatusCode::CONFLICT, + format!("catalog object already exists: {}/{}/{}", bucket, namespace.public_name(), entry.table), + )); + } + store + .ensure_table_warehouse_location_available(&entry) + .await + .map_err(catalog_store_already_exists_error)?; + return Ok(create_table_response_from_entry(entry, metadata, true)); + } ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket) .await @@ -5465,7 +5641,7 @@ where .register_table_with_publication(entry.clone(), metadata_backend) .await .map_err(catalog_store_already_exists_error)?; - Ok(load_table_response_from_entry(entry, metadata)) + Ok(create_table_response_from_entry(entry, metadata, false)) } async fn create_view_response( @@ -6080,11 +6256,14 @@ async fn standard_commit_table_response( where S: crate::table_catalog::TableCatalogStore + ?Sized, { - let Some(current) = store + let current = store .load_table(bucket, &namespace.public_name(), table) .await - .map_err(catalog_store_error)? - else { + .map_err(catalog_store_error)?; + let Some(current) = current else { + if request_has_assert_create_requirement(&request) { + return commit_staged_table_create_response(store, metadata_backend, bucket, namespace, table, request).await; + } return Err(iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_TABLE, StatusCode::NOT_FOUND, "table not found")); }; if let Some(response) = @@ -6194,6 +6373,104 @@ where Ok(commit_table_response_from_result(result, next_metadata)) } +async fn commit_staged_table_create_response( + store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, + request: RestCommitTableRequest, +) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + validate_create_table_commit_requirements(&request.requirements)?; + get_namespace_response(store, bucket, namespace).await?; + if store + .load_view(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + .is_some() + { + return Err(iceberg_rest_error( + ICEBERG_ERROR_ALREADY_EXISTS, + StatusCode::CONFLICT, + format!("catalog object already exists: {}/{}/{}", bucket, namespace.public_name(), table), + )); + } + let commit_timestamp_ms = current_time_millis(); + let metadata = apply_table_create_updates_at(&request.updates, commit_timestamp_ms)?; + validate_metadata_table_location_in_bucket(bucket, &metadata)?; + + let table = crate::table_catalog::IdentifierSegment::parse(table.to_string()) + .map_err(|err| S3Error::from(ApiError::invalid_request(format!("invalid table name: {err}"))))?; + let (commit_id, _) = standard_commit_ids(request.commit_id.or(request.idempotency_key)); + let table_id = Uuid::new_v4().to_string(); + let metadata_location = + crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id)); + let properties = serde_json::from_value::>( + metadata + .get("properties") + .cloned() + .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())), + ) + .map_err(|_| S3Error::from(ApiError::invalid_request("table property values must be strings")))?; + let mut entry = crate::table_catalog::TableEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + table_id, + table_uuid: Uuid::new_v4().to_string(), + format: "ICEBERG".to_string(), + format_version: 1, + warehouse_location: metadata_table_location(&metadata)?.to_string(), + metadata_location, + version_token: format!("token-{}", Uuid::new_v4()), + generation: 1, + state: crate::table_catalog::TableCatalogEntryState::Active, + properties, + created_at: None, + updated_at: None, + }; + adopt_registered_metadata_identity(&mut entry, &metadata)?; + validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?; + + crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket) + .await + .map_err(catalog_store_error)?; + if !crate::table_catalog::TableCommitPublication::holds_table_bucket(metadata_backend, bucket) { + return Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::Internal( + "staged table creation requires a table-bucket publication fence".to_string(), + ))); + } + let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend); + let metadata_data = serde_json::to_vec(&metadata).map_err(|err| { + S3Error::with_message(S3ErrorCode::InternalError, format!("failed to serialize initial table metadata: {err}")) + })?; + metadata_backend + .put_object( + bucket, + &entry.metadata_location, + metadata_data, + crate::table_catalog::TableCatalogPutPrecondition::IfAbsent, + ) + .await + .map_err(catalog_store_already_exists_error)?; + store + .register_table_with_publication(entry.clone(), metadata_backend) + .await + .map_err(catalog_store_already_exists_error)?; + + Ok(RestCommitTableResponse { + metadata_location: table_metadata_location_for_client(bucket, &entry.metadata_location), + metadata: table_metadata_for_client(bucket, metadata), + version_token: entry.version_token, + generation: entry.generation, + commit_id, + }) +} + async fn table_commit_for_retry( store: &S, bucket: &str, diff --git a/rustfs/src/admin/handlers/table_catalog/table.rs b/rustfs/src/admin/handlers/table_catalog/table.rs index 95e1773be..4320157d7 100644 --- a/rustfs/src/admin/handlers/table_catalog/table.rs +++ b/rustfs/src/admin/handlers/table_catalog/table.rs @@ -203,6 +203,10 @@ impl Operation for RestCommitTableHandler { install_table_catalog_s3_request_info(&mut req, &principal)?; ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?; let request = read_rest_commit_table_request(std::mem::take(&mut req.input)).await?; + if request_has_assert_create_requirement(&request) { + let namespace_resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &namespace_resource, AdminAction::CreateTableAction).await?; + } let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?; let store = table_catalog_store_from_backend(metadata_backend.clone())?; let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req); diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 4792b44fe..bb7f409e5 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -668,6 +668,11 @@ fn table_catalog_handlers_require_table_admin_actions() { ); } + let commit_table_block = operation_block(&src, "RestCommitTableHandler"); + assert!(commit_table_block.contains("request_has_assert_create_requirement(&request)")); + assert!(commit_table_block.contains("TableCatalogResource::namespace(&warehouse, &namespace)")); + assert!(commit_table_block.contains("AdminAction::CreateTableAction")); + let sync_bridge_block = operation_block(&src, "SyncExternalCatalogBridgeHandler"); assert!( sync_bridge_block.contains("AdminAction::RegisterTableAction"), @@ -2825,7 +2830,7 @@ async fn commit_request_readers_require_standard_arrays_and_preserve_legacy_poin } #[test] -fn unsupported_create_and_register_modes_return_iceberg_errors() { +fn register_overwrite_returns_iceberg_unsupported_error() { let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); let register_error = table_entry_from_register_request( "warehouse", @@ -2839,17 +2844,6 @@ fn unsupported_create_and_register_modes_return_iceberg_errors() { .expect_err("register overwrite should remain unsupported"); assert_eq!(register_error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into())); assert_eq!(register_error.status_code(), Some(StatusCode::NOT_ACCEPTABLE)); - - let create_request: CreateTableRequest = serde_json::from_value(serde_json::json!({ - "name": "events", - "schema": {"type": "struct", "schema-id": 0, "fields": []}, - "stage-create": true - })) - .expect("stage-create request should parse"); - let create_error = table_entry_from_create_table_request("warehouse", &namespace, create_request) - .expect_err("staged create should remain unsupported"); - assert_eq!(create_error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into())); - assert_eq!(create_error.status_code(), Some(StatusCode::NOT_ACCEPTABLE)); } #[test] @@ -2973,10 +2967,10 @@ async fn create_table_response_writes_initial_metadata_for_standard_request() { .expect("table should exist"); assert_eq!( response.metadata_location, - format!( + Some(format!( "s3://warehouse/.rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001-{}.metadata.json", entry.table_id - ) + )) ); assert_eq!(response.metadata["table-uuid"], entry.table_uuid); assert!( @@ -3380,7 +3374,7 @@ async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct let winner_entry = &tables[0]; assert_eq!( winner.metadata_location, - table_metadata_location_for_client("warehouse", &winner_entry.metadata_location) + Some(table_metadata_location_for_client("warehouse", &winner_entry.metadata_location)) ); let metadata_prefix = winner_entry .metadata_location @@ -3415,6 +3409,423 @@ async fn concurrent_create_table_responses_keep_one_catalog_winner_with_distinct assert!(table_uuids.contains(&winner_entry.table_uuid)); } +#[tokio::test] +async fn staged_create_is_invisible_until_assert_create_commit_across_backings() { + for mode in [ + crate::table_catalog::TableCatalogBackingMode::ObjectBacked, + crate::table_catalog::TableCatalogBackingMode::DurableStrong, + ] { + let metadata_backend = TestTableCatalogObjectBackend::content_addressed(); + let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(metadata_backend.clone(), mode); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); + + let staged = + create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true) + .await + .expect("stage-create should return initialized metadata"); + + assert_eq!(staged.metadata_location, None, "{mode:?}"); + let staged_json = serde_json::to_value(&staged).expect("staged response should serialize"); + assert!(staged_json["metadata-location"].is_null(), "{mode:?}"); + assert!( + store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .is_none(), + "{mode:?}" + ); + let metadata_prefix = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/"; + assert!( + metadata_backend + .list_objects("warehouse", metadata_prefix) + .await + .expect("metadata listing should succeed") + .is_empty(), + "{mode:?}" + ); + + let committed = commit_table_response( + &store, + &commit_backend, + "warehouse", + &namespace, + "events", + staged_create_commit_request(&staged.metadata, []), + ) + .await + .expect("assert-create commit should publish the table"); + + assert_eq!(committed.generation, 1, "{mode:?}"); + assert_eq!(committed.metadata["table-uuid"], staged.metadata["table-uuid"], "{mode:?}"); + assert_eq!(committed.metadata["metadata-log"], serde_json::json!([]), "{mode:?}"); + let entry = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("committed table should be visible"); + assert_eq!(entry.table_uuid, staged.metadata["table-uuid"].as_str().unwrap(), "{mode:?}"); + assert_eq!( + committed.metadata_location, + table_metadata_location_for_client("warehouse", &entry.metadata_location), + "{mode:?}" + ); + assert!( + metadata_backend + .object_exists("warehouse", &entry.metadata_location) + .await + .expect("metadata lookup should succeed"), + "{mode:?}" + ); + } +} + +#[tokio::test] +async fn staged_create_commit_accepts_initial_snapshot_for_spark_ctas() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::content_addressed(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); + let staged = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true) + .await + .expect("stage-create should succeed"); + let table_location = staged.metadata["location"] + .as_str() + .expect("table location should be present"); + 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 committed = commit_table_response( + &store, + &commit_backend, + "warehouse", + &namespace, + "events", + staged_create_commit_request( + &staged.metadata, + [ + serde_json::json!({ + "action": "add-snapshot", + "snapshot": { + "snapshot-id": 10, + "sequence-number": 1, + "timestamp-ms": 1234, + "manifest-list": manifest_list, + "summary": {"operation": "append"} + } + }), + serde_json::json!({ + "action": "set-snapshot-ref", + "ref-name": "main", + "snapshot-id": 10, + "type": "branch" + }), + ], + ), + ) + .await + .expect("Spark CTAS-style create commit should succeed"); + + assert_eq!(committed.metadata["current-snapshot-id"], 10); + assert_eq!(committed.metadata["last-sequence-number"], 1); + assert_eq!(committed.metadata["refs"]["main"]["snapshot-id"], 10); +} + +#[test] +fn staged_create_commit_defaults_to_format_version_two_without_upgrade() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let (_, staged_metadata) = table_entry_from_create_table_request("warehouse", &namespace, staged_events_create_request()) + .expect("staged create metadata should initialize"); + let mut request = staged_create_commit_request(&staged_metadata, []); + request + .updates + .retain(|update| update.get("action").and_then(serde_json::Value::as_str) != Some("upgrade-format-version")); + + let metadata = apply_table_create_updates_at(&request.updates, 1234) + .expect("create updates without an explicit upgrade should use the Iceberg default"); + + assert_eq!(metadata["format-version"], 2); + assert_eq!(metadata["last-sequence-number"], 0); +} + +#[tokio::test] +async fn staged_create_rejects_an_active_table_warehouse_location() { + for mode in [ + crate::table_catalog::TableCatalogBackingMode::ObjectBacked, + crate::table_catalog::TableCatalogBackingMode::DurableStrong, + ] { + let catalog_backend = TestTableCatalogObjectBackend::content_addressed(); + let metadata_backend = TestTableCatalogObjectBackend::content_addressed(); + let store = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(catalog_backend, mode); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let commit_backend = TableCommitObjectBackend::trusted(metadata_backend); + let active_request = serde_json::from_value::(serde_json::json!({ + "name": "active_events", + "location": "s3://warehouse/shared/events", + "schema": {"type": "struct", "fields": []} + })) + .expect("active create request should parse"); + create_table_response(&store, &commit_backend, "warehouse", &namespace, active_request, true) + .await + .expect("active table should be created"); + let staged_request = serde_json::from_value::(serde_json::json!({ + "name": "staged_events", + "location": "s3://warehouse/shared/events/child", + "schema": {"type": "struct", "fields": []}, + "stage-create": true + })) + .expect("staged create request should parse"); + + let error = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_request, true) + .await + .expect_err("stage-create must reject a location owned by an active table"); + + assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_ALREADY_EXISTS.into()), "{mode:?}"); + assert_eq!(error.status_code(), Some(StatusCode::CONFLICT), "{mode:?}"); + assert!( + store + .load_table("warehouse", "analytics", "staged_events") + .await + .expect("table lookup should succeed") + .is_none(), + "{mode:?}" + ); + } +} + +#[tokio::test] +async fn concurrent_staged_create_commits_publish_exactly_one_table() { + for mode in [ + crate::table_catalog::TableCatalogBackingMode::ObjectBacked, + crate::table_catalog::TableCatalogBackingMode::DurableStrong, + ] { + let catalog_backend = TestTableCatalogObjectBackend::content_addressed(); + let barrier = Arc::new(tokio::sync::Barrier::new(2)); + let metadata_backend = TestTableCatalogObjectBackend { + put_object_barrier: Some(Arc::clone(&barrier)), + ..TestTableCatalogObjectBackend::content_addressed() + }; + let store = Arc::new(crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(catalog_backend, mode)); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(store.as_ref(), "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + store.as_ref(), + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let staged = create_table_response( + store.as_ref(), + &TableCommitObjectBackend::trusted(metadata_backend.clone()), + "warehouse", + &namespace, + staged_events_create_request(), + true, + ) + .await + .expect("stage-create should succeed"); + let first_store = Arc::clone(&store); + let first_namespace = namespace.clone(); + let first_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); + let first_request = staged_create_commit_request(&staged.metadata, []); + let first = tokio::spawn(async move { + commit_table_response( + first_store.as_ref(), + &first_backend, + "warehouse", + &first_namespace, + "events", + first_request, + ) + .await + }); + tokio::time::timeout(StdDuration::from_secs(2), async { + while metadata_backend.state.lock().await.objects.is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .expect("first metadata write should reach the publication pause"); + metadata_backend.lock_attempts.lock().await.clear(); + let second_store = Arc::clone(&store); + let second_namespace = namespace.clone(); + let second_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); + let second_request = staged_create_commit_request(&staged.metadata, []); + let second = tokio::spawn(async move { + commit_table_response( + second_store.as_ref(), + &second_backend, + "warehouse", + &second_namespace, + "events", + second_request, + ) + .await + }); + metadata_backend.wait_for_lock_attempts(1).await; + assert!( + !second.is_finished(), + "second create must wait after observing the table as absent: {mode:?}" + ); + + barrier.wait().await; + let _winner = tokio::time::timeout(StdDuration::from_secs(2), first) + .await + .expect("first assert-create commit should complete") + .expect("first assert-create task should join") + .expect("first assert-create commit should win"); + tokio::time::timeout(StdDuration::from_secs(2), async { + while metadata_backend.state.lock().await.objects.len() < 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("second metadata write should reach the publication pause"); + barrier.wait().await; + let loser = tokio::time::timeout(StdDuration::from_secs(2), second) + .await + .expect("second assert-create commit should complete") + .expect("second assert-create task should join") + .expect_err("second assert-create commit must lose at atomic registration"); + + assert_eq!(loser.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_ALREADY_EXISTS.into()), "{mode:?}"); + assert_eq!(loser.status_code(), Some(StatusCode::CONFLICT), "{mode:?}"); + assert_eq!( + store + .list_tables("warehouse", "analytics") + .await + .expect("table listing should succeed") + .len(), + 1, + "{mode:?}" + ); + } +} + +#[tokio::test] +async fn malformed_or_conflicting_assert_create_does_not_replace_a_table() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::content_addressed(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); + let malformed = serde_json::from_value(serde_json::json!({ + "requirements": [{"type": "assert-create"}], + "updates": [] + })) + .expect("malformed create commit should parse"); + let error = commit_table_response(&store, &commit_backend, "warehouse", &namespace, "events", malformed) + .await + .expect_err("incomplete create updates must fail"); + assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST)); + assert!( + store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .is_none() + ); + + let staged = create_table_response(&store, &commit_backend, "warehouse", &namespace, staged_events_create_request(), true) + .await + .expect("stage-create should succeed"); + let first = commit_table_response( + &store, + &commit_backend, + "warehouse", + &namespace, + "events", + staged_create_commit_request(&staged.metadata, []), + ) + .await + .expect("first assert-create should succeed"); + let error = commit_table_response( + &store, + &commit_backend, + "warehouse", + &namespace, + "events", + staged_create_commit_request(&staged.metadata, []), + ) + .await + .expect_err("assert-create must fail after the table exists"); + assert_eq!(error.status_code(), Some(StatusCode::CONFLICT)); + let current = store + .load_table("warehouse", "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should remain visible"); + assert_eq!(current.generation, first.generation); + assert_eq!( + table_metadata_location_for_client("warehouse", ¤t.metadata_location), + first.metadata_location + ); +} + #[tokio::test] async fn standard_commit_applies_updates_and_writes_next_metadata() { let store = TestTableCatalogStore::default(); @@ -10868,6 +11279,68 @@ async fn seed_test_manifest_data_files( } } +fn staged_events_create_request() -> CreateTableRequest { + serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 1, "name": "id", "required": true, "type": "long"}, + {"id": 2, "name": "payload", "required": false, "type": "string"} + ] + }, + "stage-create": true, + "properties": {"write.format.default": "parquet"} + })) + .expect("staged create table request should parse") +} + +fn staged_create_commit_request( + staged_metadata: &serde_json::Value, + additional_updates: impl IntoIterator, +) -> RestCommitTableRequest { + let mut updates = vec![ + serde_json::json!({ + "action": "assign-uuid", + "uuid": staged_metadata["table-uuid"] + }), + serde_json::json!({ + "action": "upgrade-format-version", + "format-version": staged_metadata["format-version"] + }), + serde_json::json!({ + "action": "add-schema", + "schema": staged_metadata["schemas"][0] + }), + serde_json::json!({"action": "set-current-schema", "schema-id": -1}), + serde_json::json!({ + "action": "add-spec", + "spec": staged_metadata["partition-specs"][0] + }), + serde_json::json!({"action": "set-default-spec", "spec-id": -1}), + serde_json::json!({ + "action": "add-sort-order", + "sort-order": staged_metadata["sort-orders"][0] + }), + serde_json::json!({"action": "set-default-sort-order", "sort-order-id": -1}), + serde_json::json!({ + "action": "set-location", + "location": staged_metadata["location"] + }), + serde_json::json!({ + "action": "set-properties", + "updates": staged_metadata["properties"] + }), + ]; + updates.extend(additional_updates); + serde_json::from_value(serde_json::json!({ + "requirements": [{"type": "assert-create"}], + "updates": updates + })) + .expect("staged create commit request should parse") +} + async fn create_standard_events_table( store: &S, metadata_backend: &TestTableCatalogObjectBackend, @@ -10907,9 +11380,17 @@ where })) .expect("standard create table request should parse"); let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone()); - create_table_response(store, &commit_backend, "warehouse", namespace, create_request, true) + let response = create_table_response(store, &commit_backend, "warehouse", namespace, create_request, true) .await - .expect("table should be created") + .expect("table should be created"); + RestLoadTableResponse { + metadata_location: response + .metadata_location + .expect("direct create should return metadata location"), + metadata: response.metadata, + config: response.config, + storage_credentials: response.storage_credentials, + } } async fn create_standard_recent_events_view( diff --git a/rustfs/src/connect/diagnostics/perf_object.rs b/rustfs/src/connect/diagnostics/perf_object.rs index dda98f8ea..70aab597e 100644 --- a/rustfs/src/connect/diagnostics/perf_object.rs +++ b/rustfs/src/connect/diagnostics/perf_object.rs @@ -64,6 +64,7 @@ const SIGNATURE_PATH: &str = "envelope.sig"; const RESULT_PATH: &str = "result.json"; const MAX_OBJECT_RESPONSE_BYTES: usize = 16_384; const CLEANUP_RESERVE_MAX: Duration = Duration::from_millis(250); +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(2); const OUTPUT_MODE: u32 = 0o600; static OBJECT_COLLECTOR_ACTIVE: AtomicBool = AtomicBool::new(false); @@ -374,7 +375,7 @@ impl S3ObjectProbe { let mut builder = Client::builder() .no_proxy() .redirect(reqwest::redirect::Policy::none()) - .timeout(timeout); + .timeout(timeout.max(CLEANUP_TIMEOUT)); if let Some(root_ca_pem) = root_ca_pem { let certificate = reqwest::Certificate::from_pem(root_ca_pem).map_err(|_| ObjectPerformanceError::InvalidRootCertificate)?; @@ -403,7 +404,6 @@ impl S3ObjectProbe { let started = Instant::now(); let reserve = CLEANUP_RESERVE_MAX.min(request.duration / 10); let operation_deadline = started + request.duration.saturating_sub(reserve); - let cleanup_deadline = started + request.duration; let bucket = format!("rustfs-connect-perf-{}", request.artifact_uid.replace('-', "")); let object = "synthetic-object"; let bucket_url = self.object_url(&bucket, None)?; @@ -427,6 +427,9 @@ impl S3ObjectProbe { } Err(error) => Err(error), }; + // Cleanup owns two sequential requests and must not inherit only the + // fractional tail of the caller's measurement window. + let cleanup_deadline = Instant::now() + CLEANUP_TIMEOUT; let cleanup = self.cleanup(object_url, bucket_url, cleanup_deadline).await; if cleanup.is_err() { return Err(ObjectProbeError::CleanupFailed); diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index dd39449c6..460b69290 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -245,6 +245,26 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult>; + /// Checks whether an active table already owns any part of the candidate warehouse location. + /// + /// This is a preflight for clients that write data before catalog registration. Registration + /// remains the authoritative atomic check for concurrent creators. + async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> { + let candidate_prefix = table_warehouse_object_prefix(candidate)?; + for existing in self.list_all_tables(&candidate.table_bucket).await? { + if existing.state != TableCatalogEntryState::Active || existing.table_id == candidate.table_id { + continue; + } + let existing_prefix = table_warehouse_object_prefix(&existing)?; + if warehouse_object_prefixes_overlap(&existing_prefix, &candidate_prefix) { + return Err(TableCatalogStoreError::Conflict(format!( + "table warehouse location overlaps an active table: {candidate_prefix}" + ))); + } + } + Ok(()) + } + async fn list_tables_page( &self, table_bucket: &str, @@ -1242,6 +1262,13 @@ where } } + async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> { + match self { + Self::ObjectBacked(store) => store.ensure_table_warehouse_location_available(candidate).await, + Self::DurableStrong(store) => store.ensure_table_warehouse_location_available(candidate).await, + } + } + async fn list_tables_page( &self, table_bucket: &str, diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index a0501ab5a..6895d175a 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -5168,6 +5168,10 @@ where Ok(entries) } + async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> { + self.ensure_table_warehouse_prefix_available(candidate).await + } + async fn list_tables_page( &self, table_bucket: &str, diff --git a/rustfs/src/table_catalog/store/strong.rs b/rustfs/src/table_catalog/store/strong.rs index 455e40d64..06fde8cbc 100644 --- a/rustfs/src/table_catalog/store/strong.rs +++ b/rustfs/src/table_catalog/store/strong.rs @@ -2270,6 +2270,16 @@ where .collect()) } + async fn ensure_table_warehouse_location_available(&self, candidate: &TableEntry) -> TableCatalogStoreResult<()> { + validate_table_entry_version_and_id(candidate)?; + let namespace = parse_namespace_for_store(&candidate.namespace)?; + let table = parse_table_for_store(&candidate.table)?; + self.hydrate_state().await?; + let state = self.state.lock().await; + let key = Self::table_key(&candidate.table_bucket, &namespace, &table); + Self::ensure_table_warehouse_prefix_available_locked(&state, candidate, &key) + } + async fn list_tables_page( &self, table_bucket: &str, diff --git a/rustfs/tests/connect_perf_object.rs b/rustfs/tests/connect_perf_object.rs index 8f242e9f7..d9b6deb91 100644 --- a/rustfs/tests/connect_perf_object.rs +++ b/rustfs/tests/connect_perf_object.rs @@ -396,6 +396,61 @@ async fn oversized_response_is_rejected_without_buffering_it() { server.await.expect("response server"); } +#[tokio::test] +async fn cleanup_uses_a_budget_independent_of_the_measurement_window() { + use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _}; + + let _guard = TEST_LOCK.lock().await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("response listener"); + let address = listener.local_addr().expect("listener address"); + let server = tokio::spawn(async move { + for (index, response) in [ + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".as_slice(), + b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\n\xa5\xa5\xa5\xa5".as_slice(), + b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(), + b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n".as_slice(), + ] + .into_iter() + .enumerate() + { + let (mut socket, _) = listener.accept().await.expect("object connection"); + let mut request = vec![0_u8; 16 * 1024]; + let _ = socket.read(&mut request).await.expect("request headers"); + if index >= 3 { + tokio::time::sleep(Duration::from_millis(600)).await; + } + socket.write_all(response).await.expect("response"); + } + }); + let probe = S3ObjectProbe::new( + &format!("http://{address}"), + None, + None, + Zeroizing::new("access".to_owned()), + Zeroizing::new("secret".to_owned()), + Zeroizing::new(String::new()), + Duration::from_secs(1), + ) + .expect("object probe"); + let mut request = request(ObjectOperation::GetObject); + request.traffic_bytes = 4; + + let measurement = measure_object(&request, &probe, &CancellationToken::new()) + .await + .expect("object measurement"); + assert_eq!( + measurement.result.outcome(), + ObjectOutcome::Succeeded, + "target result: {:?}", + measurement.target + ); + tokio::time::timeout(Duration::from_secs(3), server) + .await + .expect("cleanup requests should complete") + .expect("response server"); +} + #[tokio::test] async fn signed_result_is_saved_without_overwrite() { let _guard = TEST_LOCK.lock().await; @@ -463,6 +518,7 @@ fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put() { } async fn real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put_body() { + let _guard = TEST_LOCK.lock().await; let port = match find_available_port() { Ok(port) => port, Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 2a11d17e8..aef93b8c9 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -649,8 +649,8 @@ python3 scripts/table-catalog/engine_compatibility.py \ --cleanup ``` -The generated SQL covers namespace creation, table creation, append, refresh, -count, and optional cleanup. Until Spark execution is enabled in CI through the +The generated SQL covers namespace creation, atomic CTAS through the Iceberg +REST staged-create flow, refresh, count, and optional cleanup. Until Spark execution is enabled in CI through the explicit live-conformance gate, do not claim Spark support beyond a manually verified run with the exact RustFS build, Spark version, Iceberg version, and expected output recorded. diff --git a/scripts/table-catalog/duckdb_smoke.py b/scripts/table-catalog/duckdb_smoke.py index 236cc26fe..5b89ce399 100755 --- a/scripts/table-catalog/duckdb_smoke.py +++ b/scripts/table-catalog/duckdb_smoke.py @@ -287,17 +287,6 @@ def negative_sql(args: argparse.Namespace, *, kind: str, seed_table: str, write_ bootstrap = f"bootstrap_{kind}" sql = profile_sql(args, catalog=bootstrap, table=seed_table) sql += f"DETACH {engine_compatibility.quote_double_identifier(bootstrap)};\n" - if kind == "stage-create": - catalog = "stage_default" - sql += attach_sql( - args, - catalog=catalog, - rest_path="/iceberg", - signing_name="s3", - compatibility_options=False, - ) - sql += f"CREATE TABLE {table_identifier(catalog, args.namespace, table_name(args.table, 'stage'))} (id BIGINT);\n" - return sql if kind == "purge": catalog = "purge_requested" sql += attach_sql( @@ -435,9 +424,8 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du write_table = table_name(args.table, "write") purge_table = table_name(args.table, "purge") drop_table = table_name(args.table, "drop") - stage_table = table_name(args.table, "stage") v3_table = table_name(args.table, "v3") - smoke_tables = [seed_table, write_table, purge_table, drop_table, stage_table, v3_table] + smoke_tables = [seed_table, write_table, purge_table, drop_table, v3_table] catalog = pyiceberg_smoke.load_rest_catalog(iceberg_args, deps) namespace_preexisting = bool(catalog.namespace_exists(args.namespace)) prepare_smoke_tables(catalog, args.namespace, smoke_tables, args.replace) @@ -489,7 +477,6 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du checks["s3tables_alias"] = "pass" negatives = [ - ("stage-create", "stage-create is not supported"), ("purge", "purgeRequested=true is not supported"), ("format-v3", "unsupported Iceberg table format-version: 3"), ] @@ -507,7 +494,7 @@ def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> Du ) require_duckdb_error(execution, kind, expected_error) checks[kind] = "failed-closed" - if catalog.table_exists((args.namespace, stage_table)) or catalog.table_exists((args.namespace, v3_table)): + if catalog.table_exists((args.namespace, v3_table)): raise RuntimeError("a failed DuckDB create probe left a catalog table behind") if not catalog.table_exists((args.namespace, purge_table)): raise RuntimeError("purgeRequested=true removed a table despite the expected failure") diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index e394a660e..5b7de88c0 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/scripts/table-catalog/engine_compatibility.py @@ -136,8 +136,8 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]: "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance", "scenarios": [ scenario("create-namespace", "manual-live-harness", "CREATE NAMESPACE IF NOT EXISTS"), - scenario("create-table", "manual-live-harness", "CREATE TABLE USING iceberg"), - scenario("append", "manual-live-harness", "INSERT INTO"), + scenario("create-table", "manual-live-harness", "CREATE TABLE USING iceberg AS SELECT"), + scenario("staged-create", "manual-live-harness", "atomic CTAS through the REST staged-create flow"), scenario("reload-table", "manual-live-harness", "REFRESH TABLE and SELECT COUNT"), scenario("drop-table", "manual-live-harness", "DROP TABLE and optional DROP NAMESPACE"), scenario("commit-conflict", "manual-validation-required", "requires a two-writer Spark or REST conflict harness"), @@ -162,7 +162,7 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]: scenario("catalog-attach", "automated", "attach `/iceberg` with s3 signing and `/_iceberg` with s3tables signing"), scenario("read-table", "automated", "read a PyIceberg-created table through the attached catalog"), scenario("write-table", "automated", "exercise single-table DDL, DML, schema evolution, snapshots, and PyIceberg cross-read"), - scenario("unsupported-boundaries", "automated", "verify staged create, purge, and format v3 fail closed"), + scenario("unsupported-boundaries", "automated", "verify purge and format v3 fail closed"), scenario("multi-table-mode", "automated", "verify DuckDB can avoid the multi-table commit endpoint without claiming cross-table atomicity"), ], }, @@ -363,8 +363,8 @@ def spark_sql_smoke( statements = [ f"CREATE NAMESPACE IF NOT EXISTS {namespace_identifier};", f"DROP TABLE IF EXISTS {table_identifier};", - f"CREATE TABLE {table_identifier} (id BIGINT, payload STRING) USING iceberg;", - f"INSERT INTO {table_identifier} VALUES (1, 'alpha'), (2, 'beta');", + f"CREATE TABLE {table_identifier} USING iceberg AS " + "SELECT * FROM VALUES (1L, 'alpha'), (2L, 'beta') AS source(id, payload);", f"REFRESH TABLE {table_identifier};", f"SELECT COUNT(*) AS row_count FROM {table_identifier};", ] @@ -665,7 +665,7 @@ def live_conformance_evidence( OrderedDict( [ ("client", "Spark Iceberg REST catalog"), - ("scenario", "create-namespace-create-table-append-refresh-count-cleanup"), + ("scenario", "create-namespace-staged-ctas-refresh-count-cleanup"), ("expected_status", "pass"), ("expected_row_count", 2), ("claim_after_pass", "manual-live-verified"), diff --git a/scripts/table-catalog/test_duckdb_smoke.py b/scripts/table-catalog/test_duckdb_smoke.py index cba18393f..768a95e8a 100644 --- a/scripts/table-catalog/test_duckdb_smoke.py +++ b/scripts/table-catalog/test_duckdb_smoke.py @@ -145,17 +145,6 @@ class DuckDBSmokeTest(unittest.TestCase): def test_boundary_sql_records_required_compatibility_options(self) -> None: args = self.args() - stage_sql = duckdb_smoke.negative_sql( - args, - kind="stage-create", - seed_table="events_seed", - write_table="events_write", - purge_table="events_purge", - ) - stage_attach = stage_sql.split('DETACH "bootstrap_stage-create";', 1)[1] - self.assertNotIn("STAGE_CREATE_TABLES false", stage_attach) - self.assertIn("CREATE TABLE", stage_attach) - purge_sql = duckdb_smoke.negative_sql( args, kind="purge", diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 2577e2b21..65a766f40 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -32,7 +32,7 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertEqual(spark["status"], "manual-live-harness") self.assertContainsScenario(spark, "create-namespace", "manual-live-harness") self.assertContainsScenario(spark, "create-table", "manual-live-harness") - self.assertContainsScenario(spark, "append", "manual-live-harness") + self.assertContainsScenario(spark, "staged-create", "manual-live-harness") self.assertContainsScenario(spark, "reload-table", "manual-live-harness") self.assertContainsScenario(spark, "drop-table", "manual-live-harness") self.assertContainsScenario(spark, "commit-conflict", "manual-validation-required") @@ -142,7 +142,7 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertEqual(config["spark.sql.catalog.rustfs.s3.secret-access-key"], "rustfsadmin") self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables") - def test_spark_sql_smoke_covers_lifecycle_append_reload_and_cleanup(self) -> None: + def test_spark_sql_smoke_covers_staged_ctas_reload_and_cleanup(self) -> None: sql = engine_compatibility.spark_sql_smoke( catalog_name="rustfs", namespace="sales", @@ -152,8 +152,9 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertIn("CREATE NAMESPACE IF NOT EXISTS rustfs.`sales`", sql) self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql) - self.assertIn("CREATE TABLE rustfs.`sales`.`orders`", sql) - self.assertIn("INSERT INTO rustfs.`sales`.`orders`", sql) + self.assertIn("CREATE TABLE rustfs.`sales`.`orders` USING iceberg AS", sql) + self.assertIn("SELECT * FROM VALUES (1L, 'alpha'), (2L, 'beta')", sql) + self.assertNotIn("INSERT INTO rustfs.`sales`.`orders`", sql) self.assertIn("REFRESH TABLE rustfs.`sales`.`orders`", sql) self.assertIn("SELECT COUNT(*) AS row_count FROM rustfs.`sales`.`orders`", sql) self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql) @@ -547,7 +548,7 @@ class EngineCompatibilityTest(unittest.TestCase): record = engine_compatibility.live_conformance_evidence_record( client_name="Spark Iceberg REST catalog", client_version="3.5.4/iceberg-1.7.1", - scenario="create-namespace-create-table-append-refresh-count-cleanup", + scenario="create-namespace-staged-ctas-refresh-count-cleanup", rustfs_build="rustfs-test", git_sha="abc123", catalog_backing="durable-strong",