gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-21 18:13:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tables): support staged table creation (#7919)

Browse Source
This commit is contained in:
GatewayJ
2026-09-17 06:38:41 +08:00
committed by GitHub
parent 14081c41a4
commit e3be247b40
16 changed files with 938 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/store/heal.rs
+11 -1
View File
@@ -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()
},
)
crates/ecstore/src/store/init.rs
+2 -9
View File
@@ -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));
docs/architecture/s3-tables-support-matrix.md
+3 -3
View File
@@ -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. |
rustfs/src/admin/handlers/table_catalog/mod.rs
+302 -25
View File
@@ -910,6 +910,16 @@ struct RestLoadTableResponse {
storage_credentials: Vec<RestStorageCredential>,
}
#[derive(Debug, Serialize)]
struct RestCreateTableResponse {
#[serde(rename = "metadata-location")]
metadata_location: Option<String>,
metadata: serde_json::Value,
config: BTreeMap<String, String>,
#[serde(rename = "storage-credentials")]
storage_credentials: Vec<RestStorageCredential>,
}
#[derive(Debug, Serialize)]
struct RestLoadCredentialsResponse {
config: BTreeMap<String, String>,
@@ -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<serde_json::Value> {
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<serde_json::Value> {
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<i64> {
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<serde_json::Value> {
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<i64>,
) -> S3Result<Option<i64>> {
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<i64>, label: &str) -> S3Result<
Ok(assigned_id)
}
fn next_array_object_i64(metadata: &serde_json::Value, array_key: &str, id_key: &str) -> S3Result<i64> {
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<i64>) -> S3Result<i64> {
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<i64> {
@@ -5435,11 +5582,40 @@ async fn create_table_response<S>(
namespace: &crate::table_catalog::Namespace,
request: CreateTableRequest,
table_bucket_enabled: bool,
) -> S3Result<RestLoadTableResponse>
) -> S3Result<RestCreateTableResponse>
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<S>(
@@ -6080,11 +6256,14 @@ async fn standard_commit_table_response<S>(
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<S>(
store: &S,
metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &str,
request: RestCommitTableRequest,
) -> S3Result<RestCommitTableResponse>
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::<BTreeMap<String, String>>(
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<S>(
store: &S,
bucket: &str,
rustfs/src/admin/handlers/table_catalog/table.rs
+4
View File
@@ -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);
rustfs/src/admin/handlers/table_catalog/tests.rs
+498 -17
View File
@@ -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::<CreateTableRequest>(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::<CreateTableRequest>(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", &current.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<Item = serde_json::Value>,
) -> 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<S>(
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<S>(
rustfs/src/connect/diagnostics/perf_object.rs
+5 -2
View File
@@ -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);
rustfs/src/table_catalog/store/mod.rs
+27
View File
@@ -245,6 +245,26 @@ pub(crate) trait TableCatalogStore: Send + Sync {
async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
/// 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,
rustfs/src/table_catalog/store/object.rs
+4
View File
@@ -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,
rustfs/src/table_catalog/store/strong.rs
+10
View File
@@ -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,
rustfs/tests/connect_perf_object.rs
+56
View File
@@ -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,
scripts/table-catalog/README.md
+2 -2
View File
@@ -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.
scripts/table-catalog/duckdb_smoke.py
+2 -15
View File
@@ -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")
scripts/table-catalog/engine_compatibility.py
+6 -6
View File
@@ -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"),
scripts/table-catalog/test_duckdb_smoke.py
-11
View File
@@ -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",
scripts/table-catalog/test_engine_compatibility.py
+6 -5
View File
@@ -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",