mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 18:08:21 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ea40a41c9b |
@@ -668,60 +668,6 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
|
||||
parts_metadata: &[FileInfo],
|
||||
disks: &[Option<DiskStore>],
|
||||
) -> Option<&'static str> {
|
||||
if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
|
||||
return Some(reason);
|
||||
}
|
||||
|
||||
let Ok(erasure) = coding::Erasure::try_new_with_options(
|
||||
candidate.erasure.data_blocks,
|
||||
candidate.erasure.parity_blocks,
|
||||
candidate.erasure.block_size,
|
||||
candidate.uses_legacy_checksum,
|
||||
) else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
|
||||
};
|
||||
let data_files =
|
||||
match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
|
||||
disks.get(index).is_some_and(Option::is_some)
|
||||
}) {
|
||||
Ok(data_files) => data_files,
|
||||
Err(reason) => return Some(reason),
|
||||
};
|
||||
|
||||
let Some(part) = candidate.parts.first() else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
|
||||
};
|
||||
let Ok(object_size) = usize::try_from(candidate.size) else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
|
||||
};
|
||||
let checksum_info = candidate.erasure.get_checksum_info(part.number);
|
||||
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let read_length = inline_erasure_shard_file_offset(
|
||||
0,
|
||||
object_size,
|
||||
object_size,
|
||||
candidate.erasure.block_size,
|
||||
erasure.data_shards,
|
||||
candidate.uses_legacy_checksum,
|
||||
);
|
||||
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
|
||||
let Ok(mut readers) =
|
||||
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
|
||||
else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
|
||||
};
|
||||
|
||||
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
|
||||
Some(body) if body.len() == object_size => None,
|
||||
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
|
||||
}
|
||||
}
|
||||
|
||||
fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
|
||||
// `inline_data` excludes remote objects; this diagnostic reports them separately.
|
||||
if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
|
||||
@@ -759,7 +705,51 @@ fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Op
|
||||
if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
|
||||
}
|
||||
None
|
||||
|
||||
let Ok(erasure) = coding::Erasure::try_new_with_options(
|
||||
candidate.erasure.data_blocks,
|
||||
candidate.erasure.parity_blocks,
|
||||
candidate.erasure.block_size,
|
||||
candidate.uses_legacy_checksum,
|
||||
) else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
|
||||
};
|
||||
let data_files =
|
||||
match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
|
||||
disks.get(index).is_some_and(Option::is_some)
|
||||
}) {
|
||||
Ok(data_files) => data_files,
|
||||
Err(reason) => return Some(reason),
|
||||
};
|
||||
|
||||
let Some(part) = candidate.parts.first() else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
|
||||
};
|
||||
let checksum_info = candidate.erasure.get_checksum_info(part.number);
|
||||
let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
|
||||
HashAlgorithm::HighwayHash256SLegacy
|
||||
} else {
|
||||
checksum_info.algorithm
|
||||
};
|
||||
let read_length = inline_erasure_shard_file_offset(
|
||||
0,
|
||||
object_size,
|
||||
object_size,
|
||||
candidate.erasure.block_size,
|
||||
erasure.data_shards,
|
||||
candidate.uses_legacy_checksum,
|
||||
);
|
||||
let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
|
||||
let Ok(mut readers) =
|
||||
build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
|
||||
else {
|
||||
return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
|
||||
};
|
||||
|
||||
match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
|
||||
Some(body) if body.len() == object_size => None,
|
||||
_ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
|
||||
}
|
||||
}
|
||||
|
||||
fn data_read_inline_missing_shards_are_pending(
|
||||
@@ -2620,14 +2610,6 @@ impl SetDisks {
|
||||
Ok(file_info) => {
|
||||
observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
|
||||
accumulator.observe_file_info(&file_info);
|
||||
if bounded_fanout
|
||||
&& read_data
|
||||
&& !force_full_wait
|
||||
&& let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
|
||||
{
|
||||
force_full_wait = true;
|
||||
final_miss_reason_override.get_or_insert(reason);
|
||||
}
|
||||
if let Some(slot) = ress.get_mut(index) {
|
||||
*slot = file_info;
|
||||
}
|
||||
@@ -3021,11 +3003,14 @@ impl SetDisks {
|
||||
});
|
||||
}
|
||||
|
||||
let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
|
||||
let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
|
||||
Ok(collected) => collected,
|
||||
Err(()) => return empty_quorum_result(),
|
||||
};
|
||||
|
||||
// debug!("ReadMultipleResp ress {:?}", ress);
|
||||
// debug!("ReadMultipleResp errors {:?}", errors);
|
||||
|
||||
let mut ret = Vec::with_capacity(req.files.len());
|
||||
|
||||
for want in req.files.iter() {
|
||||
@@ -7240,7 +7225,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
|
||||
async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
|
||||
const DISKS: usize = 4;
|
||||
let bucket = "bounded-data-get-hedge-bucket";
|
||||
let object = "bounded-data-get-hedge-object";
|
||||
@@ -7271,7 +7256,7 @@ mod tests {
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
|
||||
.expect("bounded data-read fanout should hedge by starting the spare disk");
|
||||
|
||||
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
|
||||
assert!(
|
||||
@@ -7287,7 +7272,7 @@ mod tests {
|
||||
assert_eq!(
|
||||
calls.total(disk_call_counters::KIND_READ_VERSION),
|
||||
DISKS as u64,
|
||||
"bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
|
||||
"bounded data-read fanout should issue the paused disk plus one spare hedge"
|
||||
);
|
||||
assert_eq!(diagnostics.total_responses(), DISKS);
|
||||
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
|
||||
@@ -7314,42 +7299,16 @@ mod tests {
|
||||
("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
|
||||
],
|
||||
async {
|
||||
let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
|
||||
let calls = disk_call_counters::observe(object);
|
||||
let disks_for_read = disks.clone();
|
||||
let mut read = tokio::spawn(async move {
|
||||
SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
|
||||
let (parts_metadata, errs, diagnostics) =
|
||||
SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
|
||||
.await
|
||||
});
|
||||
|
||||
tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
|
||||
.await
|
||||
.expect("default bounded non-inline read should schedule the paused metadata task");
|
||||
tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
|
||||
while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect(
|
||||
"default bounded non-inline read should immediately force full fanout after the first non-inline response",
|
||||
);
|
||||
|
||||
let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
|
||||
assert!(
|
||||
pending.is_err(),
|
||||
"default non-inline data reads must not return before the paused metadata response"
|
||||
);
|
||||
barrier.release();
|
||||
let (parts_metadata, errs, diagnostics) = read
|
||||
.await
|
||||
.expect("metadata read task should not panic")
|
||||
.expect("default data-read metadata should resolve");
|
||||
.expect("default data-read metadata should resolve");
|
||||
|
||||
assert_eq!(
|
||||
calls.total(disk_call_counters::KIND_READ_VERSION),
|
||||
DISKS as u64,
|
||||
"default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
|
||||
"default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
|
||||
);
|
||||
assert_eq!(diagnostics.total_responses(), DISKS);
|
||||
assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
|
||||
|
||||
@@ -714,7 +714,7 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
|
||||
const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
|
||||
|
||||
const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
|
||||
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
|
||||
const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
|
||||
|
||||
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
|
||||
const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
|
||||
@@ -1125,10 +1125,7 @@ mod prepared_get_object_metadata_tests {
|
||||
|
||||
assert_eq!(object_size, payload.len() as i64);
|
||||
assert_eq!(restored, payload);
|
||||
assert_eq!(
|
||||
calls_total, 4,
|
||||
"default production inline GET should schedule the initial bounded quorum plus one hedge"
|
||||
);
|
||||
assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
|
||||
assert_eq!(
|
||||
recorder.histogram_values(
|
||||
"rustfs_io_get_object_metadata_fanout_scheduled",
|
||||
|
||||
@@ -453,9 +453,7 @@ impl SetDisks {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// Bound, not `_`: this guard must live to the end of the scope. A bare
|
||||
// `_` would drop it here and release the namespace write lock.
|
||||
let _write_lock_guard = if !opts.no_lock {
|
||||
let write_lock_guard = if !opts.no_lock {
|
||||
let ns_lock = self.new_ns_lock(bucket, object).await?;
|
||||
Some(
|
||||
ns_lock
|
||||
@@ -998,7 +996,7 @@ impl SetDisks {
|
||||
readers.push(None);
|
||||
continue;
|
||||
}
|
||||
Err(_e) => {
|
||||
Err(e) => {
|
||||
readers.push(None);
|
||||
continue;
|
||||
}
|
||||
@@ -1547,9 +1545,6 @@ impl SetDisks {
|
||||
|
||||
for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
|
||||
for (disk_index, disk) in disks.iter().enumerate() {
|
||||
// Only the #[cfg(test)] fault-injection branch below reads this.
|
||||
#[cfg(not(test))]
|
||||
let _ = disk_index;
|
||||
let Some(disk) = disk else {
|
||||
return Ok(DanglingDeleteSafety::UnsafeToDelete);
|
||||
};
|
||||
|
||||
@@ -1400,7 +1400,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
|
||||
let mut count = max_parts;
|
||||
|
||||
for part in object_parts.iter() {
|
||||
for (i, part) in object_parts.iter().enumerate() {
|
||||
if let Some(err) = &part.error {
|
||||
warn!("list_object_parts part error: {:?}", &err);
|
||||
}
|
||||
@@ -2043,8 +2043,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
&& let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
|
||||
{
|
||||
error!(
|
||||
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
|
||||
p.part_num, bucket, object, err
|
||||
"complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
|
||||
p.part_num, bucket, object
|
||||
);
|
||||
return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
|
||||
}
|
||||
@@ -2089,8 +2089,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
|
||||
}
|
||||
} else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
|
||||
error!(
|
||||
"complete_multipart_upload checksum matches failed want={}, got={}, err={}",
|
||||
wtcs.encoded, checksum.encoded, err
|
||||
"complete_multipart_upload checksum matches failed want={}, got={}",
|
||||
wtcs.encoded, checksum.encoded
|
||||
);
|
||||
return Err(Error::other(format!(
|
||||
"complete_multipart_upload checksum matches failed want={}, got={}",
|
||||
|
||||
@@ -5656,9 +5656,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
// TODO: Lifecycle
|
||||
|
||||
let mut version_found = true;
|
||||
// delete_object_version below derives its own majority quorum from the
|
||||
// disk array, so the object-derived quorum here is unused.
|
||||
let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
|
||||
let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
|
||||
if let Some(err) = &gerr
|
||||
&& goi.name.is_empty()
|
||||
{
|
||||
@@ -6412,7 +6410,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
|
||||
self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
|
||||
|
||||
for disk in disks.iter() {
|
||||
if disk.is_some() {
|
||||
if let Some(disk) = disk {
|
||||
continue;
|
||||
}
|
||||
let _ = self
|
||||
|
||||
@@ -3937,7 +3937,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
|
||||
fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
|
||||
@@ -3946,20 +3946,20 @@ mod tests {
|
||||
],
|
||||
|| {
|
||||
assert!(is_get_metadata_data_read_early_stop_enabled());
|
||||
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
},
|
||||
);
|
||||
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
|
||||
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
|
||||
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
});
|
||||
temp_env::with_vars(
|
||||
[
|
||||
(ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
|
||||
(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
|
||||
],
|
||||
|| {
|
||||
assert!(!is_get_metadata_data_read_early_stop_enabled());
|
||||
assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
@@ -38,7 +38,6 @@ use std::collections::HashMap;
|
||||
/// - Account format is invalid
|
||||
/// - Credentials don't contain project_id
|
||||
/// - Account project_id doesn't match credentials project_id
|
||||
#[allow(dead_code)] // Used by Swift implementation
|
||||
pub fn validate_account_access(account: &str, credentials: &Credentials) -> SwiftResult<String> {
|
||||
// Extract project_id from account (strip "AUTH_" prefix)
|
||||
let account_project_id = account
|
||||
@@ -70,7 +69,6 @@ pub fn validate_account_access(account: &str, credentials: &Credentials) -> Swif
|
||||
///
|
||||
/// Admin users (with "admin" or "reseller_admin" roles) can perform
|
||||
/// cross-tenant operations and administrative tasks.
|
||||
#[allow(dead_code)] // Used by Swift implementation
|
||||
pub fn is_admin_user(credentials: &Credentials) -> bool {
|
||||
credentials
|
||||
.claims
|
||||
|
||||
@@ -144,7 +144,6 @@ impl ContainerMapper {
|
||||
/// - S3 bucket name compatible (only uses [a-z0-9-])
|
||||
/// - Deterministic mapping (same input always produces same bucket name)
|
||||
/// - Fixed-length prefix (16 hex chars = 8 bytes)
|
||||
#[allow(dead_code)] // Used in: create/delete container operations
|
||||
pub fn swift_to_s3_bucket(&self, container: &str, project_id: &str) -> String {
|
||||
if self.config.tenant_prefix_enabled {
|
||||
let hash = self.hash_project_id(project_id);
|
||||
@@ -216,7 +215,6 @@ pub fn bucket_info_to_container(info: &BucketInfo, mapper: &ContainerMapper, pro
|
||||
/// 2. Lists all S3 buckets
|
||||
/// 3. Filters to buckets belonging to this tenant (using tenant prefix)
|
||||
/// 4. Converts BucketInfo to Swift Container format
|
||||
#[allow(dead_code)] // Used by handler: list containers
|
||||
pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftResult<Vec<Container>> {
|
||||
// Validate account access and extract project_id
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -279,7 +277,6 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
|
||||
/// - Returns 201 Created on success
|
||||
/// - Returns 202 Accepted if container already exists
|
||||
/// - Returns 400 Bad Request for invalid container names
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn create_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<bool> {
|
||||
// Validate account access and extract project_id
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -348,7 +345,6 @@ fn validate_container_name(container: &str) -> SwiftResult<()> {
|
||||
}
|
||||
|
||||
/// Container metadata for HEAD response
|
||||
#[allow(dead_code)] // TODO: Remove once Swift API integration is complete
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ContainerMetadata {
|
||||
/// Number of objects in container
|
||||
@@ -411,7 +407,6 @@ pub(crate) async fn get_container_custom_metadata(
|
||||
/// - HEAD /v1/{account}/{container} returns container metadata
|
||||
/// - Returns 204 No Content on success with headers
|
||||
/// - Returns 404 Not Found if container doesn't exist
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
|
||||
let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
|
||||
|
||||
@@ -448,7 +443,6 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
|
||||
/// - The update is additive: items the request does not name keep their stored
|
||||
/// value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
|
||||
/// empty value
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn update_container_metadata(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -520,7 +514,6 @@ pub async fn update_container_metadata(
|
||||
/// - Returns 204 No Content on success
|
||||
/// - Returns 404 Not Found if container doesn't exist
|
||||
/// - Returns 409 Conflict if container is not empty
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn delete_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
|
||||
// Validate account access and extract project_id
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -603,7 +596,6 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
|
||||
/// - Account validation fails
|
||||
/// - Container doesn't exist
|
||||
/// - Storage layer errors occur
|
||||
#[allow(dead_code)] // Handler integration: GET container
|
||||
pub async fn list_objects(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -706,7 +698,6 @@ pub async fn list_objects(
|
||||
/// Versioning configuration is stored as an S3 bucket tag:
|
||||
/// - Tag key: `swift-versions-location`
|
||||
/// - Tag value: archive container name
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn enable_versioning(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -795,7 +786,6 @@ pub async fn enable_versioning(
|
||||
/// * `account` - Account identifier
|
||||
/// * `container` - Container name to disable versioning on
|
||||
/// * `credentials` - Keystone credentials
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn disable_versioning(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
|
||||
// Validate account access
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -855,7 +845,6 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
|
||||
/// # Returns
|
||||
/// - Some(archive_container_name) if versioning is enabled
|
||||
/// - None if versioning is not enabled
|
||||
#[allow(dead_code)] // Used by handler and object.rs
|
||||
pub async fn get_versions_location(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<Option<String>> {
|
||||
// Validate account access
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -918,7 +907,6 @@ pub async fn get_versions_location(account: &str, container: &str, credentials:
|
||||
/// &credentials
|
||||
/// ).await?;
|
||||
/// ```
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn set_container_acl(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -1022,7 +1010,6 @@ pub async fn set_container_acl(
|
||||
/// println!("Container is publicly readable");
|
||||
/// }
|
||||
/// ```
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn get_container_acl(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -1083,7 +1070,6 @@ pub async fn get_container_acl(
|
||||
///
|
||||
/// # Returns
|
||||
/// Ok(()) if ACLs were deleted successfully
|
||||
#[allow(dead_code)] // Used by handler
|
||||
pub async fn delete_container_acl(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
|
||||
// Setting both ACLs to None removes them
|
||||
set_container_acl(account, container, None, None, credentials).await
|
||||
|
||||
@@ -20,7 +20,6 @@ use std::fmt;
|
||||
|
||||
/// Swift-specific error type
|
||||
#[derive(Debug)]
|
||||
#[allow(dead_code)] // Error variants used by Swift implementation
|
||||
pub enum SwiftError {
|
||||
/// 400 Bad Request
|
||||
BadRequest(String),
|
||||
|
||||
@@ -122,12 +122,10 @@ fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
|
||||
///
|
||||
/// Handles URL encoding/decoding and path normalization for Swift object keys.
|
||||
/// Swift object names can contain any UTF-8 characters except null bytes.
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub struct ObjectKeyMapper;
|
||||
|
||||
impl ObjectKeyMapper {
|
||||
/// Create a new object key mapper
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn new() -> Self {
|
||||
Self
|
||||
}
|
||||
@@ -140,7 +138,6 @@ impl ObjectKeyMapper {
|
||||
/// - Not contain null bytes
|
||||
/// - Not contain '..' path segments (directory traversal)
|
||||
/// - Not start with '/' (leading slash handled by routing)
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn validate_object_name(object: &str) -> SwiftResult<()> {
|
||||
if object.is_empty() {
|
||||
return Err(SwiftError::BadRequest("Object name cannot be empty".to_string()));
|
||||
@@ -183,7 +180,6 @@ impl ObjectKeyMapper {
|
||||
/// Example:
|
||||
/// - Swift: "photos/vacation/beach photo.jpg"
|
||||
/// - S3: "photos/vacation/beach photo.jpg"
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn swift_to_s3_key(object: &str) -> SwiftResult<String> {
|
||||
Self::validate_object_name(object)?;
|
||||
Ok(object.to_string())
|
||||
@@ -193,7 +189,6 @@ impl ObjectKeyMapper {
|
||||
///
|
||||
/// This is essentially an identity transformation since we store
|
||||
/// Swift object names as-is in S3.
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn s3_to_swift_name(key: &str) -> String {
|
||||
key.to_string()
|
||||
}
|
||||
@@ -208,7 +203,6 @@ impl ObjectKeyMapper {
|
||||
/// - Object: "vacation/beach.jpg"
|
||||
/// - Bucket: "abc123:photos"
|
||||
/// - Key: "vacation/beach.jpg"
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn build_s3_key(object: &str) -> SwiftResult<String> {
|
||||
Self::swift_to_s3_key(object)
|
||||
}
|
||||
@@ -220,7 +214,6 @@ impl ObjectKeyMapper {
|
||||
///
|
||||
/// Example URL: /v1/AUTH_abc/container/path%2Fto%2Ffile.txt
|
||||
/// Decoded: "path/to/file.txt"
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn decode_object_from_url(encoded: &str) -> SwiftResult<String> {
|
||||
// Decode percent-encoding
|
||||
let decoded = urlencoding::decode(encoded).map_err(|e| SwiftError::BadRequest(format!("Invalid URL encoding: {}", e)))?;
|
||||
@@ -233,7 +226,6 @@ impl ObjectKeyMapper {
|
||||
///
|
||||
/// When constructing URLs (e.g., for redirect responses), we need to
|
||||
/// percent-encode object names.
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn encode_object_for_url(object: &str) -> String {
|
||||
urlencoding::encode(object).to_string()
|
||||
}
|
||||
@@ -241,7 +233,6 @@ impl ObjectKeyMapper {
|
||||
/// Check if object name represents a directory (pseudo-directory)
|
||||
///
|
||||
/// In Swift, objects ending with '/' are treated as directory markers.
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn is_directory_marker(object: &str) -> bool {
|
||||
object.ends_with('/')
|
||||
}
|
||||
@@ -250,7 +241,6 @@ impl ObjectKeyMapper {
|
||||
///
|
||||
/// Removes redundant slashes and normalizes the path while preserving
|
||||
/// trailing slashes for directory markers.
|
||||
#[allow(dead_code)] // Used in: object operations
|
||||
pub fn normalize_path(object: &str) -> String {
|
||||
// Split by '/', filter out empty segments (except if it's the end)
|
||||
let has_trailing_slash = object.ends_with('/');
|
||||
@@ -324,7 +314,6 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
|
||||
/// # Returns
|
||||
/// * `Ok(etag)` - Object ETag on success
|
||||
/// * `Err(SwiftError)` - Error if validation fails or upload fails
|
||||
#[allow(dead_code)] // Handler integration: PUT object
|
||||
pub async fn put_object<R>(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -445,7 +434,6 @@ where
|
||||
///
|
||||
/// Similar to put_object, but allows directly specifying metadata instead of extracting from headers.
|
||||
/// This is used internally for storing SLO manifests and marker objects.
|
||||
#[allow(dead_code)] // Used by SLO implementation
|
||||
pub async fn put_object_with_metadata<R>(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -549,7 +537,6 @@ where
|
||||
/// - `bytes=1000-1999` - Bytes 1000-1999
|
||||
/// - `bytes=1000-` - From byte 1000 to end
|
||||
/// - `bytes=-500` - Last 500 bytes
|
||||
#[allow(dead_code)] // Handler integration: GET object
|
||||
pub async fn get_object(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -608,7 +595,6 @@ pub async fn get_object(
|
||||
/// # Returns
|
||||
/// * `Ok(object_info)` - Object metadata (ObjectInfo)
|
||||
/// * `Err(SwiftError)` - Error if validation fails or object not found
|
||||
#[allow(dead_code)] // Handler integration: HEAD object
|
||||
pub async fn head_object(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -671,7 +657,6 @@ pub async fn head_object(
|
||||
/// # Returns
|
||||
/// * `Ok(())` - Object deleted successfully (or didn't exist)
|
||||
/// * `Err(SwiftError)` - Error if validation fails or deletion fails
|
||||
#[allow(dead_code)] // Handler integration: DELETE object
|
||||
pub async fn delete_object(account: &str, container: &str, object: &str, credentials: &Credentials) -> SwiftResult<()> {
|
||||
// 1. Validate account access and get project_id
|
||||
let project_id = validate_account_access(account, credentials)?;
|
||||
@@ -732,7 +717,6 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
|
||||
/// # Returns
|
||||
/// * `Ok(())` - Metadata updated successfully
|
||||
/// * `Err(SwiftError)` - Error if validation fails, object not found, or update fails
|
||||
#[allow(dead_code)] // Handler integration: POST object
|
||||
pub async fn update_object_metadata(
|
||||
account: &str,
|
||||
container: &str,
|
||||
@@ -846,7 +830,6 @@ pub async fn update_object_metadata(
|
||||
/// # Handler Integration Note
|
||||
/// The current handler architecture needs to be updated to pass headers through
|
||||
/// to support COPY method and X-Copy-From header detection. See handler.rs for details.
|
||||
#[allow(dead_code)] // Handler integration: COPY object
|
||||
#[allow(clippy::too_many_arguments)] // Necessary for full copy functionality
|
||||
pub async fn copy_object(
|
||||
src_account: &str,
|
||||
@@ -979,7 +962,6 @@ pub async fn copy_object(
|
||||
/// assert_eq!(container, "my-container");
|
||||
/// assert_eq!(object, "path/to/file.txt");
|
||||
/// ```
|
||||
#[allow(dead_code)] // Handler integration: COPY method
|
||||
pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, String)> {
|
||||
let destination = destination.trim_start_matches('/');
|
||||
let parts: Vec<&str> = destination.splitn(2, '/').collect();
|
||||
@@ -1013,7 +995,6 @@ pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, Strin
|
||||
/// # Returns
|
||||
/// * `Ok((container, object))` - Parsed container and object names
|
||||
/// * `Err(SwiftError)` - Error if format is invalid
|
||||
#[allow(dead_code)] // Handler integration: X-Copy-From
|
||||
pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)> {
|
||||
// Same parsing logic as Destination header
|
||||
parse_destination_header(copy_from)
|
||||
@@ -1042,7 +1023,6 @@ pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)>
|
||||
/// assert_eq!(range.start, 0);
|
||||
/// assert_eq!(range.end, 1023);
|
||||
/// ```
|
||||
#[allow(dead_code)] // Handler integration: Range header
|
||||
pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
|
||||
if !range_str.starts_with("bytes=") {
|
||||
return Err(SwiftError::BadRequest("Range header must start with 'bytes='".to_string()));
|
||||
@@ -1124,7 +1104,6 @@ pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
|
||||
/// let header = format_content_range(0, 1023, 5000);
|
||||
/// assert_eq!(header, "bytes 0-1023/5000");
|
||||
/// ```
|
||||
#[allow(dead_code)] // Handler integration: Range header
|
||||
pub fn format_content_range(start: i64, end: i64, total: i64) -> String {
|
||||
format!("bytes {}-{}/{}", start, end, total)
|
||||
}
|
||||
|
||||
@@ -50,7 +50,6 @@ pub enum SwiftRoute {
|
||||
|
||||
impl SwiftRoute {
|
||||
/// Get the account identifier from the route
|
||||
#[allow(dead_code)] // Public API for future use
|
||||
pub fn account(&self) -> &str {
|
||||
match self {
|
||||
SwiftRoute::Account { account, .. } => account,
|
||||
@@ -60,7 +59,6 @@ impl SwiftRoute {
|
||||
}
|
||||
|
||||
/// Extract project_id from account string (removes AUTH_ prefix)
|
||||
#[allow(dead_code)] // Public API for future use
|
||||
pub fn project_id(&self) -> Option<&str> {
|
||||
let account = self.account();
|
||||
ACCOUNT_PATTERN
|
||||
|
||||
@@ -19,7 +19,6 @@ use std::collections::HashMap;
|
||||
|
||||
/// Swift container metadata
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[allow(dead_code)] // Used in container listing operations
|
||||
pub struct Container {
|
||||
/// Container name
|
||||
pub name: String,
|
||||
@@ -34,7 +33,6 @@ pub struct Container {
|
||||
|
||||
/// Swift object metadata
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[allow(dead_code)] // Used in object listing operations
|
||||
pub struct Object {
|
||||
/// Object name (key)
|
||||
pub name: String,
|
||||
@@ -50,7 +48,6 @@ pub struct Object {
|
||||
|
||||
/// Swift metadata extracted from headers
|
||||
#[derive(Debug, Clone, Default)]
|
||||
#[allow(dead_code)] // Used by Swift implementation
|
||||
pub struct SwiftMetadata {
|
||||
/// Custom metadata key-value pairs (from X-Container-Meta-* or X-Object-Meta-*)
|
||||
pub metadata: HashMap<String, String>,
|
||||
|
||||
@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
|
||||
let CreateTableRequest {
|
||||
name,
|
||||
location,
|
||||
mut schema,
|
||||
mut partition_spec,
|
||||
mut write_order,
|
||||
schema,
|
||||
partition_spec,
|
||||
write_order,
|
||||
stage_create,
|
||||
mut properties,
|
||||
} = request;
|
||||
@@ -3031,9 +3031,6 @@ fn table_entry_from_create_table_request(
|
||||
let metadata_location =
|
||||
crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
|
||||
|
||||
crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
|
||||
.map_err(catalog_store_error)?;
|
||||
|
||||
let entry = crate::table_catalog::TableEntry {
|
||||
version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
|
||||
table_bucket: bucket.to_string(),
|
||||
|
||||
@@ -1873,66 +1873,6 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
|
||||
assert_eq!(request.name, "events");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_assigns_positive_ids_to_spark_schema() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
|
||||
"name": "events",
|
||||
"schema": {
|
||||
"type": "struct",
|
||||
"schema-id": 0,
|
||||
"fields": [
|
||||
{"id": 0, "name": "id", "required": false, "type": "long"},
|
||||
{"id": 1, "name": "payload", "required": false, "type": "string"}
|
||||
]
|
||||
},
|
||||
"partition-spec": {"spec-id": 0, "fields": []},
|
||||
"properties": {"owner": "spark"}
|
||||
}))
|
||||
.expect("Spark create table request should parse");
|
||||
|
||||
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
|
||||
.expect("catalog should assign positive field IDs");
|
||||
|
||||
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
|
||||
assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
|
||||
assert_eq!(metadata["last-column-id"], 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
|
||||
"name": "events",
|
||||
"schema": {
|
||||
"type": "struct",
|
||||
"identifier-field-ids": [-1],
|
||||
"fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
|
||||
},
|
||||
"partition-spec": {
|
||||
"fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
|
||||
},
|
||||
"write-order": {
|
||||
"fields": [{
|
||||
"source-id": -1,
|
||||
"transform": "identity",
|
||||
"direction": "asc",
|
||||
"null-order": "nulls-first"
|
||||
}]
|
||||
}
|
||||
}))
|
||||
.expect("create table request with a negative temporary field ID should parse");
|
||||
|
||||
let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
|
||||
.expect("catalog should replace the negative temporary field ID");
|
||||
|
||||
assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
|
||||
assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
|
||||
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
|
||||
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
|
||||
assert_eq!(metadata["last-column-id"], 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_request_honors_supported_format_version_property() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
@@ -2050,142 +1990,6 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
|
||||
assert_eq!(updated["default-sort-order-id"], 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
|
||||
"name": "events",
|
||||
"schema": {
|
||||
"type": "struct",
|
||||
"schema-id": 41,
|
||||
"identifier-field-ids": [0],
|
||||
"fields": [
|
||||
{"id": 0, "name": "id", "required": true, "type": "long"},
|
||||
{
|
||||
"id": 10,
|
||||
"name": "details",
|
||||
"required": false,
|
||||
"type": {
|
||||
"type": "struct",
|
||||
"fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": 20,
|
||||
"name": "tags",
|
||||
"required": false,
|
||||
"type": {
|
||||
"type": "list",
|
||||
"element-id": 21,
|
||||
"element-required": false,
|
||||
"element": "string"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": 30,
|
||||
"name": "attributes",
|
||||
"required": false,
|
||||
"type": {
|
||||
"type": "map",
|
||||
"key-id": 31,
|
||||
"key": "string",
|
||||
"value-id": 32,
|
||||
"value-required": false,
|
||||
"value": {
|
||||
"type": "struct",
|
||||
"fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
},
|
||||
"partition-spec": {
|
||||
"spec-id": 42,
|
||||
"fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
|
||||
},
|
||||
"write-order": {
|
||||
"order-id": 43,
|
||||
"fields": [{
|
||||
"source-id": 11,
|
||||
"transform": "identity",
|
||||
"direction": "asc",
|
||||
"null-order": "nulls-first"
|
||||
}]
|
||||
}
|
||||
}))
|
||||
.expect("create table request should parse");
|
||||
|
||||
let (_, metadata) =
|
||||
table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
|
||||
|
||||
let schema = &metadata["schemas"][0];
|
||||
assert_eq!(schema["fields"][0]["id"], 1);
|
||||
assert_eq!(schema["fields"][1]["id"], 2);
|
||||
assert_eq!(schema["fields"][2]["id"], 3);
|
||||
assert_eq!(schema["fields"][3]["id"], 4);
|
||||
assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
|
||||
assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
|
||||
assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
|
||||
assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
|
||||
assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
|
||||
assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
|
||||
assert_eq!(metadata["last-column-id"], 9);
|
||||
assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
|
||||
assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_rejects_duplicate_temporary_schema_field_ids() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
|
||||
"name": "events",
|
||||
"schema": {
|
||||
"type": "struct",
|
||||
"fields": [
|
||||
{"id": 0, "name": "id", "required": false, "type": "long"},
|
||||
{"id": 0, "name": "payload", "required": false, "type": "string"}
|
||||
]
|
||||
}
|
||||
}))
|
||||
.expect("create table request should parse");
|
||||
|
||||
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
|
||||
.expect_err("duplicate temporary field IDs must be rejected");
|
||||
|
||||
assert_eq!(error.message(), Some("duplicate create schema field id 0"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_table_rejects_excessive_schema_nesting() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
let mut field_type = serde_json::Value::from("long");
|
||||
for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
|
||||
field_type = serde_json::json!({
|
||||
"type": "list",
|
||||
"element-id": element_id,
|
||||
"element-required": false,
|
||||
"element": field_type
|
||||
});
|
||||
}
|
||||
let request = CreateTableRequest {
|
||||
name: "events".to_string(),
|
||||
location: None,
|
||||
schema: serde_json::json!({
|
||||
"type": "struct",
|
||||
"fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
|
||||
}),
|
||||
partition_spec: None,
|
||||
write_order: None,
|
||||
stage_create: false,
|
||||
properties: BTreeMap::new(),
|
||||
};
|
||||
|
||||
let error = table_entry_from_create_table_request("warehouse", &namespace, request)
|
||||
.expect_err("excessively nested create schemas must be rejected");
|
||||
|
||||
assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
|
||||
let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
|
||||
@@ -2443,11 +2247,7 @@ fn create_table_counts_collection_ids_in_last_column_id() {
|
||||
let (_, metadata) =
|
||||
table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
|
||||
|
||||
let schema = &metadata["schemas"][0];
|
||||
assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
|
||||
assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
|
||||
assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
|
||||
assert_eq!(metadata["last-column-id"], 5);
|
||||
assert_eq!(metadata["last-column-id"], 9);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -19,7 +19,6 @@ use futures::{StreamExt, TryStreamExt, stream};
|
||||
use super::super::*;
|
||||
|
||||
const ICEBERG_MAX_USER_FIELD_ID: i32 = i32::MAX - 200;
|
||||
pub(crate) const ICEBERG_MAX_SCHEMA_NESTING_DEPTH: usize = 128;
|
||||
|
||||
fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option<usize>) -> TableCatalogStoreResult<String> {
|
||||
let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix);
|
||||
@@ -1362,188 +1361,6 @@ fn validate_iceberg_schema(schema: &serde_json::Value, label: &str) -> TableCata
|
||||
Ok(validate_iceberg_schema_fields(schema, label)?.field_ids)
|
||||
}
|
||||
|
||||
pub(crate) fn assign_fresh_create_schema_ids(
|
||||
schema: &mut serde_json::Value,
|
||||
partition_spec: Option<&mut serde_json::Value>,
|
||||
sort_order: Option<&mut serde_json::Value>,
|
||||
) -> TableCatalogStoreResult<()> {
|
||||
let mut assigner = FreshCreateSchemaIdAssigner::new();
|
||||
assigner.assign_schema(schema)?;
|
||||
assigner.remap_identifier_field_ids(schema)?;
|
||||
if let Some(partition_spec) = partition_spec {
|
||||
assigner.remap_source_ids(partition_spec, "partition spec")?;
|
||||
}
|
||||
if let Some(sort_order) = sort_order {
|
||||
assigner.remap_source_ids(sort_order, "sort order")?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
struct FreshCreateSchemaIdAssigner {
|
||||
next_id: i32,
|
||||
old_to_new: BTreeMap<i32, i32>,
|
||||
}
|
||||
|
||||
impl FreshCreateSchemaIdAssigner {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
next_id: 1,
|
||||
old_to_new: BTreeMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn assign_schema(&mut self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
|
||||
let schema = schema
|
||||
.as_object_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema must be a JSON object".to_string()))?;
|
||||
if schema.get("type").and_then(serde_json::Value::as_str) != Some("struct") {
|
||||
return Err(TableCatalogStoreError::Invalid("create schema type must be struct".to_string()));
|
||||
}
|
||||
let fields = schema
|
||||
.get_mut("fields")
|
||||
.and_then(serde_json::Value::as_array_mut)
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be an array".to_string()))?;
|
||||
self.assign_struct_fields(fields, 0)
|
||||
}
|
||||
|
||||
fn assign_struct_fields(&mut self, fields: &mut [serde_json::Value], depth: usize) -> TableCatalogStoreResult<()> {
|
||||
for field in fields.iter_mut() {
|
||||
let field = field
|
||||
.as_object_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
|
||||
self.assign_object_id(field, "id", "create schema field id")?;
|
||||
}
|
||||
for field in fields {
|
||||
let field = field
|
||||
.as_object_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
|
||||
let field_type = field
|
||||
.get_mut("type")
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema field type is required".to_string()))?;
|
||||
self.assign_type_ids(field_type, depth)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn assign_type_ids(&mut self, field_type: &mut serde_json::Value, depth: usize) -> TableCatalogStoreResult<()> {
|
||||
if field_type.is_string() {
|
||||
return Ok(());
|
||||
}
|
||||
if depth >= ICEBERG_MAX_SCHEMA_NESTING_DEPTH {
|
||||
return Err(TableCatalogStoreError::Invalid(
|
||||
"create schema exceeds the maximum nesting depth".to_string(),
|
||||
));
|
||||
}
|
||||
let nested_depth = depth + 1;
|
||||
let field_type = field_type.as_object_mut().ok_or_else(|| {
|
||||
TableCatalogStoreError::Invalid("create schema field type must be a string or JSON object".to_string())
|
||||
})?;
|
||||
match field_type.get("type").and_then(serde_json::Value::as_str) {
|
||||
Some("struct") => {
|
||||
let fields = field_type
|
||||
.get_mut("fields")
|
||||
.and_then(serde_json::Value::as_array_mut)
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema struct fields must be an array".to_string()))?;
|
||||
self.assign_struct_fields(fields, nested_depth)
|
||||
}
|
||||
Some("list") => {
|
||||
self.assign_object_id(field_type, "element-id", "create schema list element-id")?;
|
||||
let element = field_type
|
||||
.get_mut("element")
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema list element is required".to_string()))?;
|
||||
self.assign_type_ids(element, nested_depth)
|
||||
}
|
||||
Some("map") => {
|
||||
self.assign_object_id(field_type, "key-id", "create schema map key-id")?;
|
||||
self.assign_object_id(field_type, "value-id", "create schema map value-id")?;
|
||||
let key = field_type
|
||||
.get_mut("key")
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema map key is required".to_string()))?;
|
||||
self.assign_type_ids(key, nested_depth)?;
|
||||
let value = field_type
|
||||
.get_mut("value")
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema map value is required".to_string()))?;
|
||||
self.assign_type_ids(value, nested_depth)
|
||||
}
|
||||
_ => Err(TableCatalogStoreError::Invalid(
|
||||
"create schema contains an unsupported field type".to_string(),
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn assign_object_id(
|
||||
&mut self,
|
||||
object: &mut serde_json::Map<String, serde_json::Value>,
|
||||
field: &str,
|
||||
label: &str,
|
||||
) -> TableCatalogStoreResult<()> {
|
||||
let old_id = required_i32_value(object, field, label)?;
|
||||
let entry = match self.old_to_new.entry(old_id) {
|
||||
std::collections::btree_map::Entry::Occupied(_) => {
|
||||
return Err(TableCatalogStoreError::Invalid(format!("duplicate create schema field id {old_id}")));
|
||||
}
|
||||
std::collections::btree_map::Entry::Vacant(entry) => entry,
|
||||
};
|
||||
let new_id = self.next_id;
|
||||
if new_id > ICEBERG_MAX_USER_FIELD_ID {
|
||||
return Err(TableCatalogStoreError::Invalid(
|
||||
"create schema exceeds the available Iceberg field ID range".to_string(),
|
||||
));
|
||||
}
|
||||
self.next_id = new_id.checked_add(1).ok_or_else(|| {
|
||||
TableCatalogStoreError::Invalid("create schema exceeds the available Iceberg field ID range".to_string())
|
||||
})?;
|
||||
entry.insert(new_id);
|
||||
object.insert(field.to_string(), serde_json::Value::from(new_id));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remap_identifier_field_ids(&self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
|
||||
let Some(identifier_field_ids) = schema
|
||||
.as_object_mut()
|
||||
.and_then(|schema| schema.get_mut("identifier-field-ids"))
|
||||
else {
|
||||
return Ok(());
|
||||
};
|
||||
let identifier_field_ids = identifier_field_ids
|
||||
.as_array_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid("create schema identifier-field-ids must be an array".to_string()))?;
|
||||
for field_id in identifier_field_ids {
|
||||
let old_id = required_i32(field_id, "create schema identifier field id")?;
|
||||
let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
|
||||
TableCatalogStoreError::Invalid(format!(
|
||||
"create schema identifier field id {old_id} does not reference a schema field"
|
||||
))
|
||||
})?;
|
||||
*field_id = serde_json::Value::from(*new_id);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn remap_source_ids(&self, value: &mut serde_json::Value, label: &str) -> TableCatalogStoreResult<()> {
|
||||
let value = value
|
||||
.as_object_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} must be a JSON object")))?;
|
||||
let Some(fields) = value.get_mut("fields") else {
|
||||
return Ok(());
|
||||
};
|
||||
let fields = fields
|
||||
.as_array_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be an array")))?;
|
||||
for field in fields {
|
||||
let field = field
|
||||
.as_object_mut()
|
||||
.ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be JSON objects")))?;
|
||||
let old_id = required_i32_value(field, "source-id", &format!("{label} source-id"))?;
|
||||
let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
|
||||
TableCatalogStoreError::Invalid(format!("{label} source-id {old_id} does not reference the create schema"))
|
||||
})?;
|
||||
field.insert("source-id".to_string(), serde_json::Value::from(*new_id));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_iceberg_schema_fields(schema: &serde_json::Value, label: &str) -> TableCatalogStoreResult<IcebergSchemaFields> {
|
||||
let schema = schema
|
||||
.as_object()
|
||||
|
||||
Reference in New Issue
Block a user