diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md
index a2c18d6fd..87fe01a0e 100644
--- a/docs/architecture/s3-tables-support-matrix.md
+++ b/docs/architecture/s3-tables-support-matrix.md
@@ -26,7 +26,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
| `/_iceberg/v1` | Supported compatibility alias | MinIO AIStor-style alias; smoke profile defaults to signing name `s3tables`. |
| S3 object data plane | Supported | Data, metadata, manifest, and delete files are ordinary S3 objects with table-aware policy checks on warehouse paths. |
| S3 table-bucket listings | Supported | `ListObjects`, `ListObjectsV2`, `ListObjectVersions`, metadata-list extensions, and multipart-upload listings omit warehouse objects unless the caller has `admin:GetTableMetadata` for that table, and always omit internal `.rustfs-table` keys. Protected cursors are encrypted with a key derived from the cluster root credentials, so rotating those credentials invalidates cursors already issued. |
-| Object-backed warehouse-index recovery | Supported with bounded request fallback | Data-plane requests repair or verify a missing warehouse-prefix index against bounded pages totaling at most 4,096 catalog objects. Larger catalogs fail closed with a retryable service error and require `POST /iceberg/v1/{warehouse}/catalog/warehouse-index/backfill` with `admin:MigrateTableCatalog` instead of allowing one S3 request to trigger an unbounded scan. |
+| Object-backed warehouse-index recovery | Supported with bounded request fallback | Data-plane requests repair or verify a missing warehouse-prefix index by listing the catalog prefix. Only table entries count toward the 4,096-object limit, so maintenance reports and other per-table objects cannot exhaust it; the listing itself is bounded to 64 pages. Larger catalogs fail closed with a retryable service error and require operator backfill. An optional table-entry directory (see Maintenance Matrix) answers these lookups without the prefix listing. |
| Table bucket enablement | Supported | A regular bucket is enabled for catalog use and addressed as the REST catalog warehouse. |
| Catalog-vended table credentials | Automated when enabled | Disabled by default. LoadTable vends credentials only when `X-Iceberg-Access-Delegation` contains the exact `vended-credentials` token; the dedicated credentials endpoint uses the same issuer path. |
| AWS S3 Tables endpoint shape | Profile generator | Generates the AWS catalog URI and warehouse ARN shape for migration docs. API parity not claimed. |
@@ -106,12 +106,24 @@ expiration and compaction plan/commit, remains object-backed only. A configured
server has opt-in, allowlisted periodic queue admission; see the
[scheduler runbook](../operations/s3-tables-maintenance-scheduler.md).
-Renamed tables retain their original metadata paths. Durable-strong scheduling
-and cleanup fail closed with a conflict if another table shares a scanned
-metadata or warehouse directory, including a recreated source name after rename.
-Such tables require disjoint cleanup roots before maintenance can run; enabling
+Renamed tables retain their original metadata paths. Scheduling, planning,
+and cleanup on both backings fail closed with a conflict if another table
+shares a scanned metadata or warehouse directory, including a recreated source
+name after rename. The object backing rechecks this ownership inventory under
+the bucket publication fence before every delete pass; the inventory reads every
+table entry in the bucket, so maintenance cost grows with the table count. Such
+tables require disjoint cleanup roots before maintenance can run; enabling
deletion does not override this ownership check.
+Object-backed create, register, and commit publication share one preparation deadline, `RUSTFS_TABLE_CATALOG_PUBLICATION_PREPARATION_TIMEOUT_SECS` (default 30, never below the object lock-acquire timeout). Migration and bucket-lock queueing, reference HEAD validation, ownership inventories, metadata parsing, and snapshot preparation consume that same interval; repeated preparation does not renew it. A create writes its initial metadata file before registration without ending the deadline, so an orphaned file (reclaimed by maintenance) is the worst outcome of an expired create. Expiry before a possible durable catalog write returns HTTP 503 and releases the preparation locks; a valid large operation that cannot finish within the interval keeps failing on retry until the interval is raised. Once a catalog write may have begun, this deadline no longer cancels that transaction. Individual PUT completion and persisted commit/rename recovery retain their existing cancellation guarantees. This is a preparation limit, not a hard bound on disk I/O or the durable tail. The durable-strong backing does not implement this deadline; it is being retired. A rename that the client abandons mid-flight completes or recovers on a detached task, so the bucket's `active_rename_id` is always cleared.
+
+Compaction commits hold inputs to a 512 MiB stored and decoded budget per rewrite group, validate the footer and every page header before decoding a page (a recursive or over-deep Iceberg manifest Avro schema and a Parquet page that exceeds its budget are refused with HTTP 400), and run one compaction per node at a time; a concurrent request receives HTTP 503.
+
+### Table-entry directory
+
+An optional per-bucket directory of table entries under `.rustfs.sys` answers data-plane index-miss lookups without listing the catalog prefix. It is off by default: `RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX=true` and `RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX_FLEET_CONFIRMED=true` on every node, then `POST /iceberg/v1/{warehouse}/catalog/directory` (`admin:MigrateTableCatalog`) advances a backfill of 128 catalog objects per call until the directory reports `Ready`; `GET` reports the phase and `DELETE` rolls the directory back. Entry writers add a directory candidate only while the directory exists, and the backfill covers earlier entries. The directory state occupies the backing-migration fence object: nodes without this feature answer catalog mutations on that bucket with a parse error until they are upgraded, which is what the fleet-confirmed flag attests, and a backing migration is refused with a conflict while the directory is active.
+
+
Durable-strong rename preserves maintenance configuration, job identifiers,
current/latest aliases, history, audit events, and quarantine state. A persisted
rename intent protects sidecar migration across interruption: the source remains
diff --git a/rustfs/src/admin/handlers/table_catalog/config.rs b/rustfs/src/admin/handlers/table_catalog/config.rs
index 9f94d670e..47a668105 100644
--- a/rustfs/src/admin/handlers/table_catalog/config.rs
+++ b/rustfs/src/admin/handlers/table_catalog/config.rs
@@ -293,3 +293,46 @@ impl Operation for SyncExternalCatalogBridgeHandler {
build_json_response(StatusCode::OK, &response)
}
}
+
+pub struct GetTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for GetTableDirectoryHandler {
+ async fn call(&self, req: S3Request
, params: Params<'_, '_>) -> S3Result> {
+ let warehouse = warehouse_from_params(¶ms)?;
+ let resource = TableCatalogResource::warehouse(&warehouse);
+ authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableCatalogAction).await?;
+ ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+ let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+ let response = store.table_directory_status(&warehouse).await.map_err(catalog_store_error)?;
+ build_json_response(StatusCode::OK, &response)
+ }
+}
+
+pub struct AdvanceTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for AdvanceTableDirectoryHandler {
+ async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> {
+ let warehouse = warehouse_from_params(¶ms)?;
+ authorize_table_catalog_request(&req, AdminAction::MigrateTableCatalogAction).await?;
+ ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+ let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+ let response = store.advance_table_directory(&warehouse).await.map_err(catalog_store_error)?;
+ build_json_response(StatusCode::OK, &response)
+ }
+}
+
+pub struct DisableTableDirectoryHandler {}
+
+#[async_trait::async_trait]
+impl Operation for DisableTableDirectoryHandler {
+ async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> {
+ let warehouse = warehouse_from_params(¶ms)?;
+ authorize_table_catalog_request(&req, AdminAction::MigrateTableCatalogAction).await?;
+ ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+ let store = table_catalog_object_store_from_extensions(&req.extensions)?;
+ let response = store.disable_table_directory(&warehouse).await.map_err(catalog_store_error)?;
+ build_json_response(StatusCode::OK, &response)
+ }
+}
diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs
index c429835fa..e041e93aa 100644
--- a/rustfs/src/admin/handlers/table_catalog/mod.rs
+++ b/rustfs/src/admin/handlers/table_catalog/mod.rs
@@ -1364,8 +1364,14 @@ enum TableCommitObjectIdentity {
ContentSha256(String),
}
+// Legacy objects without a write-unique body identity retain SHA validation,
+// with a total byte budget independent of the manifest graph's larger limit.
+const TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES: usize = 2 * MAX_ADMIN_REQUEST_BODY_SIZE;
+
#[derive(Clone, PartialEq, Eq)]
struct TableCommitObservedObject {
+ body_revision: Option,
+ content_size: usize,
identity: TableCommitObjectIdentity,
max_size: Option,
}
@@ -1377,6 +1383,7 @@ struct TableCommitPublicationState {
table_fence: Option<(String, String, String)>,
catalog_migration_read_permit: bool,
observed_objects: BTreeMap<(String, String), TableCommitObservedObject>,
+ legacy_revalidation_bytes: usize,
guards: Vec,
}
@@ -1387,6 +1394,7 @@ struct TableCommitObjectBackend {
authorization_error: Arc>>,
publication: Arc>,
publication_fence_fleet_confirmed: bool,
+ preparation: Arc,
}
impl TableCommitObjectBackend
@@ -1400,6 +1408,7 @@ where
authorization_error: Arc::new(tokio::sync::Mutex::new(None)),
publication: Arc::new(parking_lot::Mutex::new(TableCommitPublicationState::default())),
publication_fence_fleet_confirmed,
+ preparation: Arc::new(crate::table_catalog::TablePublicationPreparation::default()),
}
}
@@ -1445,6 +1454,62 @@ where
)
}
+ async fn prepare_put_observation(
+ &self,
+ bucket: &str,
+ object: &str,
+ data: &[u8],
+ ) -> crate::table_catalog::TableCatalogStoreResult {
+ self.prepare_put_observation_with(bucket, object, data, true).await
+ }
+
+ /// `begins_mutation` is false for a write that precedes the catalog
+ /// transaction (see `put_object_before_publication`): it is still fenced
+ /// and authorized, but it leaves the preparation deadline armed.
+ async fn prepare_put_observation_with(
+ &self,
+ bucket: &str,
+ object: &str,
+ data: &[u8],
+ begins_mutation: bool,
+ ) -> crate::table_catalog::TableCatalogStoreResult {
+ let fence = {
+ let publication = self.publication.lock();
+ publication.bucket_fence.as_ref().map(|_| {
+ crate::table_catalog::TableCatalogObjectMutationFence::from_signals(
+ publication
+ .guards
+ .iter()
+ .map(crate::table_catalog::TableCatalogLockGuard::lock_lost_signal),
+ )
+ })
+ };
+ let authorize = self.authorize(bucket, object, S3Action::PutObjectAction);
+ if fence.is_some() {
+ self.preparation.read(self, authorize).await?;
+ } else {
+ authorize.await?;
+ }
+ self.ensure_observation_allowed(bucket, object)?;
+ let observation = TableCommitObservedObject {
+ identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(data, str::to_string)),
+ body_revision: None,
+ content_size: data.len(),
+ max_size: Some(data.len()),
+ };
+ if let Some(fence) = fence {
+ // An unfenced candidate write precedes the publication transaction. Only a PUT
+ // under the bucket fence commits this preparation to finishing its durable tail.
+ fence.ensure_held()?;
+ if begins_mutation {
+ self.preparation.begin_mutation()?;
+ } else {
+ self.preparation.check_deadline()?;
+ }
+ }
+ Ok(observation)
+ }
+
async fn authorize(&self, bucket: &str, object: &str, action: S3Action) -> crate::table_catalog::TableCatalogStoreResult<()> {
let result = match &self.authorization {
TableCommitObjectAuthorization::Request(req) => {
@@ -1502,31 +1567,62 @@ where
bucket: &str,
object: &str,
observation: TableCommitObservedObject,
+ enforce_legacy_budget: bool,
) -> crate::table_catalog::TableCatalogStoreResult<()> {
let key = (bucket.to_string(), object.to_string());
let mut publication = self.publication.lock();
if publication.phase == TableCommitPublicationPhase::Complete {
return Ok(());
}
+ let previous_legacy_bytes = publication.observed_objects.get(&key).map_or(0, |observed| {
+ if observed.body_revision.is_none() {
+ observed.content_size
+ } else {
+ 0
+ }
+ });
+ let legacy_bytes = publication.legacy_revalidation_bytes - previous_legacy_bytes;
+ let legacy_bytes = legacy_bytes.checked_add(if observation.body_revision.is_none() {
+ observation.content_size
+ } else {
+ 0
+ });
+ let Some(legacy_bytes) =
+ legacy_bytes.filter(|bytes| !enforce_legacy_budget || *bytes <= TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES)
+ else {
+ return Err(crate::table_catalog::TableCatalogStoreError::Unavailable(
+ "legacy catalog objects exceed the publication validation budget; rewrite the referenced objects and retry"
+ .to_string(),
+ ));
+ };
let Some(existing) = publication.observed_objects.get_mut(&key) else {
if publication.phase != TableCommitPublicationPhase::Discovering {
return Err(crate::table_catalog::TableCatalogStoreError::Internal(format!(
"table commit accessed an unvalidated object after publication preparation: {bucket}/{object}"
)));
}
+ publication.legacy_revalidation_bytes = legacy_bytes;
publication.observed_objects.insert(key, observation);
return Ok(());
};
- if existing.identity != observation.identity {
+ if existing.identity != observation.identity
+ || existing
+ .body_revision
+ .zip(observation.body_revision)
+ .is_some_and(|(before, after)| before != after)
+ {
return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
"table commit object changed during validation: {bucket}/{object}"
)));
}
+ existing.body_revision = observation.body_revision;
+ existing.content_size = observation.content_size;
existing.max_size = match (existing.max_size, observation.max_size) {
(Some(expected), Some(actual)) => Some(expected.min(actual)),
(Some(expected), None) => Some(expected),
(None, actual) => actual,
};
+ publication.legacy_revalidation_bytes = legacy_bytes;
Ok(())
}
@@ -1538,7 +1634,12 @@ where
None => TableCommitObjectIdentity::Missing,
Some(object) => TableCommitObjectIdentity::ContentSha256(hex_sha256(&object.data, str::to_string)),
};
- TableCommitObservedObject { identity, max_size }
+ TableCommitObservedObject {
+ identity,
+ max_size,
+ body_revision: object.and_then(|object| object.body_revision).filter(|id| !id.is_nil()),
+ content_size: object.map_or(0, |object| object.data.len()),
+ }
}
fn observed_metadata(
@@ -1560,10 +1661,18 @@ where
Ok(TableCommitObservedObject {
identity,
max_size: None,
+ body_revision: None,
+ content_size: 0,
})
}
async fn begin_bucket_publication(&self, table_bucket: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
+ self.preparation
+ .read(self, self.begin_bucket_publication_inner(table_bucket))
+ .await
+ }
+
+ async fn begin_bucket_publication_inner(&self, table_bucket: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
{
let mut publication = self.publication.lock();
if publication.phase != TableCommitPublicationPhase::Discovering {
@@ -1610,6 +1719,17 @@ where
table_bucket: &str,
namespace: &str,
table: &str,
+ ) -> crate::table_catalog::TableCatalogStoreResult<()> {
+ self.preparation
+ .read(self, self.prepare_publication_inner(table_bucket, namespace, table))
+ .await
+ }
+
+ async fn prepare_publication_inner(
+ &self,
+ table_bucket: &str,
+ namespace: &str,
+ table: &str,
) -> crate::table_catalog::TableCatalogStoreResult<()> {
let namespace = crate::table_catalog::Namespace::parse(namespace)
.map_err(|err| crate::table_catalog::TableCatalogStoreError::Invalid(format!("invalid namespace: {err}")))?;
@@ -1619,6 +1739,12 @@ where
let publication_lock = crate::table_catalog::default_table_publication_lock_path(&namespace, &table);
let expected = {
let mut publication = self.publication.lock();
+ if publication.legacy_revalidation_bytes > TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES {
+ return Err(crate::table_catalog::TableCatalogStoreError::Unavailable(
+ "legacy catalog objects exceed the publication validation budget; rewrite the referenced objects and retry"
+ .to_string(),
+ ));
+ }
match publication.phase {
TableCommitPublicationPhase::Prepared if publication.table_fence.as_ref() == Some(&table_fence) => {
return Ok(());
@@ -1668,10 +1794,19 @@ where
if !matches!(&expected_object.identity, TableCommitObjectIdentity::ContentSha256(_)) {
continue;
}
- let actual_object = match expected_object.max_size {
- Some(max_size) => self.backend.read_object_unlocked_limited(bucket, object, max_size).await?,
- None => self.backend.read_object_unlocked(bucket, object).await?,
- };
+ if let Some(revision) = expected_object.body_revision {
+ let actual = self.backend.object_metadata_unlocked(bucket, object).await?;
+ if actual.and_then(|metadata| metadata.body_revision) != Some(revision) {
+ return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
+ "table commit object changed before catalog publication: {bucket}/{object}"
+ )));
+ }
+ continue;
+ }
+ let max_size = expected_object
+ .max_size
+ .map_or(expected_object.content_size, |limit| limit.min(expected_object.content_size));
+ let actual_object = self.backend.read_object_unlocked_limited(bucket, object, max_size).await?;
let actual = Self::observed_content(actual_object.as_ref(), expected_object.max_size);
if actual.identity != expected_object.identity {
return Err(crate::table_catalog::TableCatalogStoreError::Conflict(format!(
@@ -1762,7 +1897,7 @@ where
self.authorize(bucket, object, S3Action::GetObjectAction).await?;
self.ensure_observation_allowed(bucket, object)?;
let result = self.backend.read_object(bucket, object).await?;
- self.record_observation(bucket, object, Self::observed_content(result.as_ref(), None))?;
+ self.record_observation(bucket, object, Self::observed_content(result.as_ref(), None), true)?;
Ok(result)
}
@@ -1775,7 +1910,7 @@ where
self.authorize(bucket, object, S3Action::GetObjectAction).await?;
self.ensure_observation_allowed(bucket, object)?;
let result = self.backend.read_object_limited(bucket, object, max_size).await?;
- self.record_observation(bucket, object, Self::observed_content(result.as_ref(), Some(max_size)))?;
+ self.record_observation(bucket, object, Self::observed_content(result.as_ref(), Some(max_size)), true)?;
Ok(result)
}
@@ -1802,7 +1937,7 @@ where
let metadata = self.backend.object_metadata(bucket, object).await?;
let observation = Self::observed_metadata(bucket, object, metadata)?;
let exists = observation.identity != TableCommitObjectIdentity::Missing;
- self.record_observation(bucket, object, observation)?;
+ self.record_observation(bucket, object, observation, true)?;
Ok(exists)
}
@@ -1817,14 +1952,24 @@ where
data: Vec,
precondition: crate::table_catalog::TableCatalogPutPrecondition,
) -> crate::table_catalog::TableCatalogStoreResult<()> {
- self.authorize(bucket, object, S3Action::PutObjectAction).await?;
- self.ensure_observation_allowed(bucket, object)?;
- let observation = TableCommitObservedObject {
- identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(&data, str::to_string)),
- max_size: None,
- };
+ let observation = self.prepare_put_observation(bucket, object, &data).await?;
self.backend.put_object(bucket, object, data, precondition).await?;
- self.record_observation(bucket, object, observation)
+ // The existing generated-metadata readback binds the written bytes to
+ // their revision. Until then, keep the original SHA expectation and
+ // enforce the fallback budget before publication preparation.
+ self.record_observation(bucket, object, observation, false)
+ }
+
+ async fn put_object_before_publication(
+ &self,
+ bucket: &str,
+ object: &str,
+ data: Vec,
+ precondition: crate::table_catalog::TableCatalogPutPrecondition,
+ ) -> crate::table_catalog::TableCatalogStoreResult<()> {
+ let observation = self.prepare_put_observation_with(bucket, object, &data, false).await?;
+ self.backend.put_object(bucket, object, data, precondition).await?;
+ self.record_observation(bucket, object, observation, false)
}
async fn put_object_unlocked(
@@ -1835,16 +1980,11 @@ where
precondition: crate::table_catalog::TableCatalogPutPrecondition,
write_guards: Vec,
) -> crate::table_catalog::TableCatalogStoreResult<()> {
- self.authorize(bucket, object, S3Action::PutObjectAction).await?;
- self.ensure_observation_allowed(bucket, object)?;
- let observation = TableCommitObservedObject {
- identity: TableCommitObjectIdentity::ContentSha256(hex_sha256(&data, str::to_string)),
- max_size: None,
- };
+ let observation = self.prepare_put_observation(bucket, object, &data).await?;
self.backend
.put_object_unlocked(bucket, object, data, precondition, write_guards)
.await?;
- self.record_observation(bucket, object, observation)
+ self.record_observation(bucket, object, observation, false)
}
async fn delete_object(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<()> {
@@ -1890,6 +2030,10 @@ where
self.backend.acquire_write_lock(bucket, object).await
}
+ fn table_publication_preparation_budget(&self) -> Arc {
+ self.preparation.clone()
+ }
+
async fn begin_table_bucket_commit_publication(
&self,
table_bucket: &str,
@@ -2522,6 +2666,29 @@ where
.map_err(catalog_store_error)
}
+/// Recheck the bucket registered before the publication lock. A missing entry
+/// here can indicate retirement while queued; do not recreate it under the fence.
+async fn require_table_bucket_for_publication(
+ store: &S,
+ bucket: &str,
+ publication: &(dyn crate::table_catalog::TableCommitPublication + Sync),
+) -> S3Result<()>
+where
+ S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
+ let entry = publication
+ .preparation_budget()
+ .read(publication, store.get_table_bucket(bucket))
+ .await
+ .map_err(catalog_store_error)?;
+ if entry.is_some_and(|entry| entry.state == crate::table_catalog::TableCatalogEntryState::Active) {
+ return Ok(());
+ }
+ Err(catalog_store_error(crate::table_catalog::TableCatalogStoreError::NotFound(format!(
+ "table bucket {bucket}"
+ ))))
+}
+
async fn table_bucket_response(store: &S, bucket: &str, enabled: bool) -> S3Result
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
@@ -6014,11 +6181,11 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
- ensure_table_bucket_entry(store, bucket, true).await?;
+ require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
let metadata_data = serde_json::to_vec(&metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize initial table metadata: {}", err))?;
metadata_backend
- .put_object(
+ .put_object_before_publication(
bucket,
&entry.metadata_location,
metadata_data,
@@ -6055,7 +6222,7 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
- ensure_table_bucket_entry(store, bucket, true).await?;
+ require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
let metadata_data = serde_json::to_vec(&metadata)
.map_err(|err| s3_error!(InternalError, "failed to serialize initial view metadata: {}", err))?;
metadata_backend
@@ -6478,15 +6645,18 @@ async fn publish_table_commit(
where
S: crate::table_catalog::TableCatalogStore + ?Sized,
{
- let _bucket_publication_completion = if table_bucket_fence_required {
- metadata_backend
- .begin_table_bucket_commit_publication(&request.table_bucket)
- .await
- .map_err(catalog_store_error)?;
- Some(crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend))
- } else {
- None
- };
+ // Real request backends acquire the migration permit here, before the store chooses its
+ // inline/child route. Nested publication reuses that same permit and preparation budget.
+ let _bucket_publication_completion =
+ if table_bucket_fence_required || metadata_backend.acquires_catalog_migration_read_permit() {
+ metadata_backend
+ .begin_table_bucket_commit_publication(&request.table_bucket)
+ .await
+ .map_err(catalog_store_error)?;
+ Some(crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend))
+ } else {
+ None
+ };
store
.commit_table_with_publication(request, metadata_backend)
.await
@@ -6843,6 +7013,7 @@ where
.map_err(catalog_store_error)?;
validate_table_metadata_snapshot_graph(metadata_backend, bucket, &entry, None, &metadata).await?;
+ ensure_table_bucket_entry(store, bucket, true).await?;
crate::table_catalog::TableCommitPublication::begin_table_bucket(metadata_backend, bucket)
.await
.map_err(catalog_store_error)?;
@@ -6852,12 +7023,12 @@ where
)));
}
let _publication_completion = crate::table_catalog::TableCommitPublicationCompletion::new(metadata_backend);
- ensure_table_bucket_entry(store, bucket, true).await?;
+ require_table_bucket_for_publication(store, bucket, metadata_backend).await?;
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(
+ .put_object_before_publication(
bucket,
&entry.metadata_location,
metadata_data,
diff --git a/rustfs/src/admin/handlers/table_catalog/routes.rs b/rustfs/src/admin/handlers/table_catalog/routes.rs
index 5ba191123..802bb75c6 100644
--- a/rustfs/src/admin/handlers/table_catalog/routes.rs
+++ b/rustfs/src/admin/handlers/table_catalog/routes.rs
@@ -39,6 +39,21 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix
format!("{prefix}/buckets/{{warehouse}}").as_str(),
AdminOperation(&GET_TABLE_BUCKET_HANDLER),
)?;
+ r.insert(
+ Method::GET,
+ format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+ AdminOperation(&GetTableDirectoryHandler {}),
+ )?;
+ r.insert(
+ Method::POST,
+ format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+ AdminOperation(&AdvanceTableDirectoryHandler {}),
+ )?;
+ r.insert(
+ Method::DELETE,
+ format!("{prefix}/{{warehouse}}/catalog/directory").as_str(),
+ AdminOperation(&DisableTableDirectoryHandler {}),
+ )?;
r.insert(
Method::GET,
format!("{prefix}/{{warehouse}}/catalog/migration").as_str(),
diff --git a/rustfs/src/admin/handlers/table_catalog/table.rs b/rustfs/src/admin/handlers/table_catalog/table.rs
index 66ea2f404..9f572d103 100644
--- a/rustfs/src/admin/handlers/table_catalog/table.rs
+++ b/rustfs/src/admin/handlers/table_catalog/table.rs
@@ -56,6 +56,25 @@ impl Operation for RestRenameTableHandler {
ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
let store = table_catalog_store_from_extensions(&req.extensions)?;
+ rename_table_response(store, warehouse, source_namespace, source_table, destination_namespace, destination_table).await?;
+ Ok(empty_response(StatusCode::NO_CONTENT))
+ }
+}
+
+// Authorization has completed before ownership moves to the mutation task. A
+// disconnected request must not strand the persisted bucket-wide rename fence.
+pub(super) async fn rename_table_response(
+ store: S,
+ warehouse: String,
+ source_namespace: crate::table_catalog::Namespace,
+ source_table: crate::table_catalog::IdentifierSegment,
+ destination_namespace: crate::table_catalog::Namespace,
+ destination_table: crate::table_catalog::IdentifierSegment,
+) -> S3Result<()>
+where
+ S: crate::table_catalog::TableCatalogStore + Send + 'static,
+{
+ crate::admin::handlers::supervise_admin_mutation("table rename", async move {
store
.rename_table(
&warehouse,
@@ -65,9 +84,9 @@ impl Operation for RestRenameTableHandler {
destination_table.as_str(),
)
.await
- .map_err(catalog_store_error)?;
- Ok(empty_response(StatusCode::NO_CONTENT))
- }
+ .map_err(catalog_store_error)
+ })
+ .await
}
pub struct RestCreateTableHandler {}
diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs
index f43154dfa..49af69533 100644
--- a/rustfs/src/admin/handlers/table_catalog/tests.rs
+++ b/rustfs/src/admin/handlers/table_catalog/tests.rs
@@ -277,6 +277,41 @@ async fn table_catalog_authentication_and_credentials_use_the_request_context()
assert!(Arc::ptr_eq(&resolved_store, &store));
let mut request = request;
+ iam.set_policy(policy_name, Policy::parse_config(br#"{
+ "Version":"2012-10-17","Statement":[
+ {"Effect":"Allow","Action":["admin:GetTable","admin:CreatePluginDelegation"]},
+ {"Effect":"Deny","Action":["admin:GetTable","admin:CreatePluginDelegation"],"Condition":{"NotIpAddress":{"aws:SourceIp":"10.0.0.0/8"}}}
+ ]
+ }"#).unwrap()).await.unwrap();
+ request
+ .extensions
+ .insert(Some(crate::server::RemoteAddr("10.1.2.3:4567".parse().unwrap())));
+ request.headers.insert("x-forwarded-for", "10.1.2.3".parse().unwrap());
+ for (client, allowed) in [
+ (None, true),
+ (Some("203.0.113.17:7890"), false),
+ (Some("10.8.7.6:7890"), true),
+ ] {
+ request.extensions.remove::();
+ if let Some(client) = client {
+ request
+ .extensions
+ .insert(rustfs_trusted_proxies::ClientInfo::direct(client.parse().unwrap()));
+ }
+ let unscoped = authorize_table_catalog_request(&request, AdminAction::GetTableAction).await;
+ let scoped = authorize_table_catalog_resource_request(&request, &resource, AdminAction::GetTableAction).await;
+ let delegated =
+ crate::admin::handlers::plugin_delegation::owner(&request, AdminAction::CreatePluginDelegationAction).await;
+ if allowed {
+ unscoped.expect("verified private address authorizes table catalog access");
+ scoped.expect("verified private address authorizes scoped table access");
+ delegated.expect("verified private address authorizes delegation");
+ } else {
+ assert_eq!(unscoped.expect_err("external client is denied").code(), &S3ErrorCode::AccessDenied);
+ assert_eq!(scoped.err().expect("external client is denied").code(), &S3ErrorCode::AccessDenied);
+ assert_eq!(delegated.err().expect("external client is denied").code(), &S3ErrorCode::AccessDenied);
+ }
+ }
request.method = Method::POST;
request.uri = "/iceberg/v1/request-warehouse/namespaces/analytics/tables/events/maintenance/worker/run"
.parse()
@@ -1473,6 +1508,7 @@ fn table_catalog_handlers_require_table_admin_actions() {
2
);
assert_eq!(rename_block.matches("AdminAction::SetTableAction").count(), 2);
+ assert!(rename_block.contains("rename_table_response("));
let migration_block = operation_block(&src, "GetTableCatalogMigrationHandler");
assert!(
@@ -1731,7 +1767,7 @@ fn table_bucket_toggle_handlers_reload_peer_metadata_after_publication() {
#[test]
fn table_commit_publication_acquires_migration_permits_before_publication_lock() {
let src = table_catalog_handler_source();
- let block = function_block(&src, "async fn begin_bucket_publication");
+ let block = function_block(&src, "async fn begin_bucket_publication_inner");
let migration_permits = block
.find("acquire_catalog_migration_read_guards(table_bucket)")
.expect("publication should acquire catalog migration permits");
@@ -7128,6 +7164,371 @@ async fn commit_publication_acquires_discovered_object_locks_in_key_order() {
crate::table_catalog::TableCatalogObjectBackend::complete_table_commit_publication(&commit_backend);
}
+#[tokio::test]
+async fn catalog_publication_reads_graph_once_for_registration_and_relocation() {
+ for relocate in [false, true] {
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+ let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace");
+ let (metadata_location, mut metadata, version_token) = if relocate {
+ let created = create_standard_events_table(&store, &backend, &namespace).await;
+ let current = store
+ .load_table("warehouse", "analytics", "events")
+ .await
+ .expect("load table")
+ .expect("table exists");
+ let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table");
+ (
+ crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002-relocation.metadata.json"),
+ created.metadata,
+ Some(current.version_token),
+ )
+ } else {
+ let (_, location) = seed_events_registration_target(&store, &backend).await;
+ (
+ location,
+ test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id"),
+ None,
+ )
+ };
+ let warehouse = "s3://warehouse/tables/revision-graph";
+ let manifest_list = format!("{warehouse}/metadata/snap-10.avro");
+ let manifest = format!("{warehouse}/metadata/manifest-snap-10.avro");
+ let data_file = format!("{warehouse}/data/part-10.parquet");
+ metadata["location"] = serde_json::json!(warehouse);
+ metadata["last-sequence-number"] = serde_json::json!(1);
+ metadata["snapshots"] = serde_json::json!([{
+ "snapshot-id": 10, "sequence-number": 1, "timestamp-ms": 10,
+ "manifest-list": manifest_list, "summary": {"operation": "append"}
+ }]);
+ metadata["current-snapshot-id"] = serde_json::json!(10);
+ metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+ backend.put_json("warehouse", &metadata_location, metadata).await;
+ seed_test_snapshot_manifest(&backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await;
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ if let Some(version_token) = version_token {
+ update_table_metadata_location_response(
+ &store,
+ &commit,
+ "warehouse",
+ &namespace,
+ "events",
+ UpdateTableMetadataLocationRequest {
+ metadata_location: table_metadata_location_for_client("warehouse", &metadata_location),
+ version_token,
+ commit_id: Some("revision-relocation".to_string()),
+ idempotency_key: None,
+ },
+ )
+ .await
+ .expect("relocation publishes the validated graph");
+ } else {
+ register_table_response(
+ &store,
+ &commit,
+ "warehouse",
+ &namespace,
+ RegisterTableRequest {
+ name: "events".to_string(),
+ metadata_location: metadata_location.clone(),
+ overwrite: false,
+ },
+ true,
+ )
+ .await
+ .expect("registration publishes the validated graph");
+ }
+ let published = store
+ .load_table("warehouse", "analytics", "events")
+ .await
+ .expect("load published table")
+ .expect("table exists");
+ assert_eq!(published.warehouse_location, warehouse);
+ assert_eq!(published.metadata_location, metadata_location);
+ let state = backend.state.lock().await;
+ for location in [&manifest_list, &manifest] {
+ let key = ("warehouse".to_string(), test_snapshot_object_key("warehouse", location));
+ assert_eq!(
+ state.read_attempts.get(&key),
+ Some(&1),
+ "manifest bodies must be read once, before the bucket fence: {location}"
+ );
+ assert_eq!(
+ state.read_lock_acquisitions.get(&key),
+ Some(&1),
+ "publication must retain the manifest object read lock"
+ );
+ }
+ }
+}
+
+#[tokio::test]
+async fn catalog_create_binds_generated_metadata_before_publication() {
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+ let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace");
+ let created = create_standard_events_table(&store, &backend, &namespace).await;
+ let object = table_metadata_location_for_catalog("warehouse", &created.metadata_location).expect("metadata key");
+ let state = backend.state.lock().await;
+ let key = ("warehouse".to_string(), object);
+ assert_eq!(
+ state.read_attempts.get(&key),
+ Some(&1),
+ "initial create keeps its generated metadata SHA read bounded"
+ );
+ assert!(
+ state
+ .objects
+ .get(&key)
+ .expect("created metadata exists")
+ .body_revision
+ .is_some()
+ );
+ assert_eq!(
+ state.read_lock_acquisitions.get(&key),
+ Some(&1),
+ "create keeps its metadata body stable until publication"
+ );
+}
+
+#[tokio::test]
+async fn commit_publication_revalidates_legacy_bytes_with_a_total_budget() {
+ use crate::table_catalog::{TableCatalogObjectBackend as _, TableCatalogStoreError};
+
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let first = "metadata/legacy-first.avro";
+ let second = "metadata/legacy-second.avro";
+ let payload = vec![b'x'; TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES / 2 + 1];
+ for object in [first, second] {
+ backend.put_bytes("warehouse", object, payload.clone()).await;
+ backend
+ .state
+ .lock()
+ .await
+ .objects
+ .get_mut(&("warehouse".to_string(), object.to_string()))
+ .expect("legacy fixture exists")
+ .body_revision = None;
+ }
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ commit
+ .read_object("warehouse", first)
+ .await
+ .expect("first legacy object fits");
+ commit
+ .read_object("warehouse", first)
+ .await
+ .expect("repeated observations do not consume the budget twice");
+ let error = commit
+ .read_object("warehouse", second)
+ .await
+ .expect_err("combined legacy payload exceeds the budget");
+ assert!(matches!(error, TableCatalogStoreError::Unavailable(_)));
+ assert!(
+ backend.lock_attempts.lock().await.is_empty(),
+ "oversized legacy validation fails before publication locks"
+ );
+
+ // A supported small legacy object still uses the original body fingerprint,
+ // even when metadata and the absent revision cannot distinguish an overwrite.
+ let legacy = "metadata/legacy-small.json";
+ backend.put_bytes("warehouse", legacy, b"before".to_vec()).await;
+ backend
+ .state
+ .lock()
+ .await
+ .objects
+ .get_mut(&("warehouse".to_string(), legacy.to_string()))
+ .expect("small legacy fixture exists")
+ .body_revision = None;
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ commit.read_object("warehouse", legacy).await.expect("read legacy object");
+ backend
+ .state
+ .lock()
+ .await
+ .objects
+ .get_mut(&("warehouse".to_string(), legacy.to_string()))
+ .expect("legacy object remains")
+ .data = b"after!".to_vec();
+ let error = commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect_err("legacy SHA validation detects an equal-length overwrite with unchanged metadata");
+ assert!(matches!(error, TableCatalogStoreError::Conflict(_)));
+
+ // Rewriting old objects supplies fresh server revisions and removes the
+ // legacy-only size restriction without removing publication validation.
+ for object in [first, second] {
+ backend.put_bytes("warehouse", object, payload.clone()).await;
+ }
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ for object in [first, second] {
+ commit
+ .read_object("warehouse", object)
+ .await
+ .expect("rewritten object can be validated");
+ }
+ let reads = backend.read_call_count().await;
+ commit
+ .begin_table_bucket_commit_publication("warehouse")
+ .await
+ .expect("take bucket fence");
+ commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect("publish rewritten objects");
+ assert_eq!(
+ backend.read_call_count().await,
+ reads,
+ "revision validation must not reread bodies under the bucket fence"
+ );
+ commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_bounds_unread_generated_metadata() {
+ use crate::table_catalog::{TableCatalogObjectBackend as _, TableCatalogPutPrecondition, TableCatalogStoreError};
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let payload = vec![b'x'; TABLE_COMMIT_LEGACY_REVALIDATION_MAX_BYTES + 1];
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ commit
+ .put_object("warehouse", "metadata/generated.json", payload, TableCatalogPutPrecondition::IfAbsent)
+ .await
+ .expect("write generated metadata before its normal readback");
+ let error = commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect_err("publication cannot read oversized unbound generated metadata under its locks");
+ assert!(matches!(error, TableCatalogStoreError::Unavailable(_)));
+ assert!(backend.lock_attempts.lock().await.is_empty());
+ assert_eq!(backend.read_call_count().await, 0);
+ commit
+ .read_object("warehouse", "metadata/generated.json")
+ .await
+ .expect("the existing generated metadata readback binds the revision");
+ let reads = backend.read_call_count().await;
+ commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect("bound generated metadata can publish");
+ assert_eq!(backend.read_call_count().await, reads);
+ commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_accepts_metadata_only_changes_with_same_body_revision() {
+ use crate::table_catalog::TableCatalogObjectBackend as _;
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ backend
+ .put_bytes("warehouse", "metadata/current.json", b"unchanged body".to_vec())
+ .await;
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ commit
+ .read_object("warehouse", "metadata/current.json")
+ .await
+ .expect("observe body");
+ {
+ let mut state = backend.state.lock().await;
+ let object = state
+ .objects
+ .get_mut(&("warehouse".to_string(), "metadata/current.json".to_string()))
+ .expect("object exists");
+ object.etag = "metadata-only-update".to_string();
+ object.mod_time = Some(time::OffsetDateTime::now_utc());
+ }
+ commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect("unchanged body remains valid after a metadata-only update");
+ assert_eq!(backend.read_call_count().await, 1);
+ commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn commit_publication_accepts_unchanged_legacy_bytes() {
+ use crate::table_catalog::TableCatalogObjectBackend as _;
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ backend
+ .put_bytes("warehouse", "metadata/legacy.json", b"unchanged".to_vec())
+ .await;
+ backend
+ .state
+ .lock()
+ .await
+ .objects
+ .get_mut(&("warehouse".to_string(), "metadata/legacy.json".to_string()))
+ .expect("legacy fixture exists")
+ .body_revision = None;
+ let commit = TableCommitObjectBackend::trusted(backend.clone());
+ commit
+ .read_object("warehouse", "metadata/legacy.json")
+ .await
+ .expect("observe legacy body");
+ commit
+ .prepare_table_commit_publication("warehouse", "analytics", "events")
+ .await
+ .expect("small legacy object remains supported");
+ assert_eq!(
+ backend.read_call_count().await,
+ 2,
+ "legacy body must still be verified under its object lock"
+ );
+ commit.complete_table_commit_publication();
+}
+
+#[tokio::test]
+async fn catalog_body_revision_changes_on_real_identical_body_overwrite() {
+ use crate::table_catalog::{
+ EcStoreTableCatalogObjectBackend, StrongTableCatalogRuntime, TableCatalogObjectBackend as _, TableCatalogPutPrecondition,
+ };
+ let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+ let bucket = "catalog-body-revision";
+ store
+ .make_bucket(bucket, &MakeBucketOptions::default())
+ .await
+ .expect("create bucket");
+ let backend = EcStoreTableCatalogObjectBackend::new_with_strong_runtime(store, StrongTableCatalogRuntime::default());
+ backend
+ .put_object(bucket, "manifest.avro", b"same bytes".to_vec(), TableCatalogPutPrecondition::IfAbsent)
+ .await
+ .expect("write object");
+ let before = backend
+ .read_object(bucket, "manifest.avro")
+ .await
+ .expect("read body")
+ .expect("object exists");
+ let before_stat = backend
+ .object_metadata(bucket, "manifest.avro")
+ .await
+ .expect("stat body")
+ .expect("object exists");
+ assert!(before.body_revision.is_some());
+ assert_eq!(before.body_revision, before_stat.body_revision);
+ backend
+ .put_object(
+ bucket,
+ "manifest.avro",
+ b"same bytes".to_vec(),
+ TableCatalogPutPrecondition::IfMatch(before.etag.clone().expect("etag exists")),
+ )
+ .await
+ .expect("overwrite same bytes");
+ let after = backend
+ .read_object(bucket, "manifest.avro")
+ .await
+ .expect("read overwritten body")
+ .expect("object exists");
+ let after_stat = backend
+ .object_metadata(bucket, "manifest.avro")
+ .await
+ .expect("stat overwritten body")
+ .expect("object exists");
+ assert_eq!(before.etag, after.etag);
+ assert_ne!(before.body_revision, after.body_revision);
+ assert_eq!(after.body_revision, after_stat.body_revision);
+}
+
#[tokio::test]
async fn commit_publication_revalidates_objects_after_ordered_lock_acquisition() {
let metadata_backend = TestTableCatalogObjectBackend::content_addressed();
@@ -7160,6 +7561,7 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
metadata_backend.state.lock().await.objects.insert(
("warehouse".to_string(), object.to_string()),
TestCatalogObjectRecord {
+ body_revision: Some(Uuid::new_v4()),
data: replacement,
etag: original_etag.clone(),
mod_time: None,
@@ -7173,6 +7575,7 @@ async fn commit_publication_binds_fingerprint_to_returned_bytes() {
metadata_backend.state.lock().await.objects.insert(
("warehouse".to_string(), object.to_string()),
TestCatalogObjectRecord {
+ body_revision: Some(Uuid::new_v4()),
data: original,
etag: original_etag,
mod_time: None,
@@ -7226,6 +7629,7 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
state.objects.insert(
("warehouse".to_string(), test_snapshot_object_key("warehouse", file)),
TestCatalogObjectRecord {
+ body_revision: Some(Uuid::new_v4()),
data: data.clone(),
etag: etag.clone(),
mod_time: None,
@@ -7256,7 +7660,7 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
]
}))
.expect("boundary commit request should parse");
- let commit_backend = TableCommitObjectBackend::trusted(metadata_backend);
+ let commit_backend = TableCommitObjectBackend::trusted(metadata_backend.clone());
let result = standard_commit_table_response(&store, &commit_backend, "warehouse", &namespace, "events", request).await;
let response = commit_backend
@@ -7265,6 +7669,14 @@ async fn standard_commit_publishes_more_than_ten_thousand_live_files() {
.expect("a valid snapshot with more than ten thousand live files should publish");
assert_eq!(response.metadata["current-snapshot-id"], 20);
+ for location in [&manifest_list, &manifest] {
+ let key = ("warehouse".to_string(), test_snapshot_object_key("warehouse", location));
+ assert_eq!(
+ metadata_backend.state.lock().await.read_attempts.get(&key),
+ Some(&2),
+ "graph and conflict validation each read manifests once; publication must not read them again"
+ );
+ }
assert_eq!(
store
.load_table("warehouse", "analytics", "events")
@@ -7565,6 +7977,78 @@ async fn metadata_location_api_rejects_unbound_legacy_catalog_identity() {
assert_events_table_entry_unchanged(&store, &legacy_entry).await;
}
+#[tokio::test]
+async fn rename_request_cancellation_finishes_the_persisted_bucket_fence() {
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+ let namespace = crate::table_catalog::Namespace::parse("analytics").unwrap();
+ let source = crate::table_catalog::IdentifierSegment::parse("events").unwrap();
+ let destination = crate::table_catalog::IdentifierSegment::parse("renamed").unwrap();
+ create_standard_events_table(&store, &backend, &namespace).await;
+ let original = store.load_table("warehouse", "analytics", "events").await.unwrap().unwrap();
+ let paths = crate::table_catalog::TableCatalogObjectPaths::default();
+ let source_path = paths.table_entry_path("warehouse", &namespace, &source);
+ let attempt = backend
+ .put_attempt_count(crate::admin::storage_api::RUSTFS_META_BUCKET, &source_path)
+ .await
+ + 2;
+ let pause = backend
+ .pause_put_attempt(crate::admin::storage_api::RUSTFS_META_BUCKET, &source_path, attempt)
+ .await;
+ let request = tokio::spawn(table::rename_table_response(
+ store.clone(),
+ "warehouse".to_string(),
+ namespace.clone(),
+ source.clone(),
+ namespace.clone(),
+ destination.clone(),
+ ));
+ tokio::time::timeout(std::time::Duration::from_secs(10), pause.wait_started())
+ .await
+ .expect("rename reaches persisted fence");
+ assert!(
+ store
+ .get_table_bucket("warehouse")
+ .await
+ .unwrap()
+ .unwrap()
+ .active_rename_id
+ .is_some()
+ );
+ request.abort();
+ assert!(request.await.unwrap_err().is_cancelled());
+ pause.release();
+ tokio::time::timeout(std::time::Duration::from_secs(10), async {
+ while store
+ .get_table_bucket("warehouse")
+ .await
+ .unwrap()
+ .unwrap()
+ .active_rename_id
+ .is_some()
+ {
+ tokio::task::yield_now().await;
+ }
+ })
+ .await
+ .expect("detached mutation clears the durable fence");
+ let renamed = store.load_table("warehouse", "analytics", "renamed").await.unwrap().unwrap();
+ assert_eq!(renamed.table_id, original.table_id);
+ assert_eq!(renamed.metadata_location, original.metadata_location);
+ assert!(store.load_table("warehouse", "analytics", "events").await.unwrap().is_none());
+ assert!(
+ store
+ .resolve_table_data_plane_resource("warehouse", "ordinary.txt")
+ .await
+ .unwrap()
+ .is_none()
+ );
+ table::rename_table_response(store.clone(), "warehouse".to_string(), namespace.clone(), destination, namespace, source)
+ .await
+ .expect("subsequent ordinary rename succeeds");
+ assert!(store.load_table("warehouse", "analytics", "events").await.unwrap().is_some());
+}
+
#[tokio::test]
async fn table_metadata_maintenance_helper_runs_dry_run_and_delete() {
let backend = TestTableCatalogObjectBackend::content_addressed();
@@ -15963,6 +16447,7 @@ struct ColdStrongRegistrationBackend {
armed: Arc,
reader_ready: Arc,
resume: Arc,
+ backup_writer_waiting: Arc,
}
impl ColdStrongRegistrationBackend {
@@ -16012,7 +16497,16 @@ impl crate::table_catalog::T
bucket: &str,
object: &str,
) -> crate::table_catalog::TableCatalogStoreResult {
- self.inner.acquire_write_lock(bucket, object).await
+ let mut acquire = self.inner.acquire_write_lock(bucket, object);
+ if bucket == crate::admin::storage_api::RUSTFS_META_BUCKET
+ && object == crate::table_catalog::TableCatalogObjectPaths::default().backing_migration_global_fence_lock_path()
+ {
+ match futures::poll!(acquire.as_mut()) {
+ std::task::Poll::Ready(result) => return result,
+ std::task::Poll::Pending => self.backup_writer_waiting.notify_one(),
+ }
+ }
+ acquire.await
}
async fn read_object(
@@ -16116,6 +16610,7 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
armed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
reader_ready: Arc::new(tokio::sync::Notify::new()),
resume: Arc::new(tokio::sync::Notify::new()),
+ backup_writer_waiting: Arc::new(tokio::sync::Notify::new()),
};
let catalog = StrongTableCatalogStore::new(backend.clone());
for bucket in [&bucket, &backup_bucket] {
@@ -16176,17 +16671,18 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
.await
.expect("registration holds first real global reader");
// The backup targets another bucket, so its documented writer-drain precondition is met.
- let backup_catalog = StrongTableCatalogStore::new(inner);
- let mut backup = Box::pin(backup_catalog.create_durable_catalog_backup(&backup_bucket, None));
- assert!(
- futures::poll!(backup.as_mut()).is_pending(),
- "actual backup writer must queue behind held registration reader"
- );
+ let backup_catalog = StrongTableCatalogStore::new(backend.clone());
+ // Keep driving route discovery and snapshot I/O until the global write lock itself is pending.
+ let backup = tokio::spawn(async move { backup_catalog.create_durable_catalog_backup(&backup_bucket, None).await });
+ tokio::time::timeout(StdDuration::from_secs(5), backend.backup_writer_waiting.notified())
+ .await
+ .expect("actual backup writer must queue behind held registration reader");
backend.resume.notify_one();
let result = tokio::time::timeout(StdDuration::from_secs(2), &mut registration).await;
if result.is_err() {
// Prove that the queued backup, rather than ECStore I/O latency, caused the stall.
- drop(backup);
+ backup.abort();
+ assert!(backup.await.unwrap_err().is_cancelled());
tokio::time::timeout(StdDuration::from_secs(5), &mut registration)
.await
.expect("registration must resume once only the queued backup is cancelled")
@@ -16201,6 +16697,7 @@ async fn cold_strong_registration_completes_while_other_bucket_backup_waits() {
tokio::time::timeout(StdDuration::from_secs(5), backup)
.await
.expect("the other bucket backup should resume when registration releases its reader")
+ .expect("backup task joins")
.expect("the other bucket backup succeeds");
}
@@ -16528,6 +17025,66 @@ async fn cleanup_empty_table_bucket_reclaims_twenty_dropped_tables_and_allows_bu
object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
}
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_empty_table_bucket_validates_directory_candidate_path_and_body_before_deletion() {
+ use crate::admin::storage_api::RUSTFS_META_BUCKET;
+ use crate::table_catalog::{TableCatalogObjectPaths, TableCatalogPutPrecondition};
+
+ let (_temp, _disks, storage) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+ let bucket = format!("cleanup-directory-{}", Uuid::new_v4().simple());
+ let backend = crate::table_catalog::EcStoreTableCatalogObjectBackend::new_with_strong_runtime(
+ storage.clone(),
+ crate::table_catalog::StrongTableCatalogRuntime::default(),
+ );
+ let catalog = crate::table_catalog::ConfiguredTableCatalogStore::new_for_test(
+ backend.clone(),
+ crate::table_catalog::TableCatalogBackingMode::ObjectBacked,
+ );
+ let metadata = create_dropped_empty_tables_for_cleanup(&catalog, &backend, &storage, &bucket, 1).await;
+ let paths = TableCatalogObjectPaths::default();
+ let bucket_entry = paths.table_bucket_entry_path(&bucket);
+ let directory = format!("{}/table-entry-directory/v1/", bucket_entry.rsplit_once('/').unwrap().0);
+ let real = format!("{directory}analytics/tables/events_00/table-entry.json");
+ // Entry writers add candidates only while the directory is active; seed the
+ // one a backfill would have written so cleanup has a valid candidate to keep.
+ let real_target = format!("{}analytics/tables/events_00/table-entry.json", paths.namespace_entries_prefix(&bucket));
+ backend
+ .put_object(RUSTFS_META_BUCKET, &real, real_target.into_bytes(), TableCatalogPutPrecondition::IfAbsent)
+ .await
+ .unwrap();
+ assert!(backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+ for (suffix, target_suffix) in [
+ ("bad segment/tables/x/table-entry.json", "bad segment/tables/x/table-entry.json"),
+ ("analytics/tables/forged/table-entry.json", "analytics/tables/events_00/table-entry.json"),
+ ] {
+ let unknown = format!("{directory}{suffix}");
+ let target = format!("{}{target_suffix}", paths.namespace_entries_prefix(&bucket));
+ backend
+ .put_object(RUSTFS_META_BUCKET, &unknown, target.into_bytes(), TableCatalogPutPrecondition::IfAbsent)
+ .await
+ .unwrap();
+ let publication = TableCommitObjectBackend::trusted(backend.clone());
+ assert!(
+ cleanup_empty_table_bucket_response(&catalog, &publication, &storage, &bucket)
+ .await
+ .is_err()
+ );
+ assert!(backend.object_exists(RUSTFS_META_BUCKET, &unknown).await.unwrap());
+ assert!(backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+ assert!(backend.object_exists(&bucket, &metadata[0]).await.unwrap());
+ backend.delete_object(RUSTFS_META_BUCKET, &unknown).await.unwrap();
+ }
+ let publication = TableCommitObjectBackend::trusted(backend.clone());
+ assert!(
+ !cleanup_empty_table_bucket_response(&catalog, &publication, &storage, &bucket)
+ .await
+ .unwrap()
+ .enabled
+ );
+ assert!(!backend.object_exists(RUSTFS_META_BUCKET, &real).await.unwrap());
+}
+
#[tokio::test]
#[serial_test::serial]
async fn cleanup_empty_table_bucket_validates_authorization_and_metadata_before_deleting() {
@@ -16705,7 +17262,7 @@ async fn cleanup_empty_table_bucket_handler_enforces_request_object_permissions(
assert!(!backend.object_exists(&bucket, object).await.unwrap());
}
} else {
- assert_eq!(result.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+ assert_eq!(result.expect_err("request must be denied").code(), &S3ErrorCode::AccessDenied);
for object in &metadata_objects {
assert!(backend.object_exists(&bucket, object).await.unwrap());
}
@@ -17099,3 +17656,202 @@ async fn cleanup_empty_table_bucket_stops_before_delete_when_publication_lease_i
);
object_store.delete_bucket(&bucket, &Default::default()).await.unwrap();
}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_first_fenced_metadata_put_finishes_after_budget() {
+ use crate::table_catalog::{TableCatalogObjectBackend, TableCatalogPutPrecondition, TableCommitPublication};
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ let publication = TableCommitObjectBackend::trusted(backend.clone());
+ publication.begin_table_bucket("warehouse").await.unwrap();
+ let key = "generated.metadata.json";
+ let pause = backend.pause_next_put("warehouse", key).await;
+ let worker = publication.clone();
+ let put = tokio::spawn(async move {
+ worker
+ .put_object("warehouse", key, b"{}".to_vec(), TableCatalogPutPrecondition::IfAbsent)
+ .await
+ });
+ pause.wait_started().await;
+ tokio::time::advance(crate::storage_api::table::get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+ assert!(publication.holds_table_bucket("warehouse"));
+ assert!(!put.is_finished());
+ pause.release();
+ put.await.unwrap().unwrap();
+ publication
+ .preparation
+ .read(&publication, backend.object_metadata("warehouse", key))
+ .await
+ .unwrap()
+ .unwrap();
+ assert!(publication.holds_table_bucket("warehouse"));
+ publication.complete();
+}
+
+#[tokio::test(start_paused = true)]
+async fn publication_deadline_slow_head_clears_partial_http_preparation() {
+ use crate::table_catalog::test_support::TestCatalogObjectPause;
+ use crate::table_catalog::{TableCatalogObjectBackend, TableCommitPublication};
+ let backend = TestTableCatalogObjectBackend::content_addressed();
+ backend.seed_object("warehouse", "data/file.parquet", b"data".to_vec()).await;
+ let publication = TableCommitObjectBackend::trusted(backend.clone());
+ publication.object_exists("warehouse", "data/file.parquet").await.unwrap();
+ publication.begin_table_bucket("warehouse").await.unwrap();
+ let pause = TestCatalogObjectPause::default();
+ backend
+ .state
+ .lock()
+ .await
+ .pause_metadata
+ .insert(("warehouse".to_string(), "data/file.parquet".to_string()), pause.clone());
+ let preparing = {
+ let publication = publication.clone();
+ tokio::spawn(async move { publication.prepare("warehouse", "analytics", "events").await })
+ };
+ pause.wait_started().await;
+ tokio::time::advance(crate::storage_api::table::get_lock_acquire_timeout() + StdDuration::from_millis(1)).await;
+ assert!(
+ matches!(preparing.await.unwrap(), Err(crate::table_catalog::TableCatalogStoreError::Unavailable(message)) if message.contains("preparation deadline"))
+ );
+ assert!(!publication.holds_table_bucket("warehouse"));
+ assert!(!publication.holds_table("warehouse", "analytics", "events"));
+ assert!(publication.publication.lock().observed_objects.is_empty());
+ assert!(backend.state.lock().await.put_attempts.is_empty());
+ pause.release();
+ let _writer = backend
+ .acquire_read_lock("warehouse", &crate::table_catalog::default_table_bucket_publication_lock_path())
+ .await
+ .unwrap();
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn directory_management_handlers_enforce_migration_permission_before_mutation() {
+ temp_env::async_with_vars([
+ (crate::table_catalog::ENV_TABLE_CATALOG_BACKING, Some("object")),
+ ("RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX", Some("true")),
+ ("RUSTFS_TABLE_CATALOG_DIRECTORY_INDEX_FLEET_CONFIRMED", Some("true")),
+ ], async {
+
+ let (_temp_dir, _disks, object_store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
+ object_store
+ .make_bucket("warehouse", &MakeBucketOptions::default())
+ .await
+ .unwrap();
+ rustfs_iam::store::object::ObjectStore::new(object_store.clone())
+ .save_iam_config(
+ serde_json::json!({"version": 1}),
+ format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+ )
+ .await
+ .unwrap();
+ let iam = rustfs_iam::build_iam_sys(object_store.clone()).await.unwrap();
+ let access_key = "directory-upgrade-user";
+ let secret_key = "directory-upgrade-secret";
+ iam.create_user(
+ access_key,
+ &AddOrUpdateUserReq {
+ secret_key: secret_key.to_string(),
+ policy: None,
+ status: AccountStatus::Enabled,
+ },
+ )
+ .await
+ .unwrap();
+ let context = Arc::new(AppContext::new(
+ object_store.clone(),
+ Arc::new(RequestIam { handle: iam.clone() }),
+ Arc::new(RequestKms),
+ ));
+ assert!(context.publish_action_credentials(rustfs_credentials::Credentials {
+ access_key: "directory-root".to_string(),
+ secret_key: "directory-root-secret".to_string(),
+ status: "on".to_string(),
+ ..Default::default()
+ }));
+ let setup = table_catalog_handler_request(context.clone(), access_key, secret_key);
+ let backend = table_catalog_backend_from_extensions(&setup.extensions).unwrap();
+ let configured = table_catalog_store_from_backend(backend.clone()).unwrap();
+ enable_table_bucket_marker(object_store.as_ref(), "warehouse").await.unwrap();
+ ensure_table_bucket_entry(&configured, "warehouse", true).await.unwrap();
+
+ for (name, policy, allowed) in [
+ (
+ "directory-maintenance",
+ serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:RunTableMaintenance"]}]}),
+ false,
+ ),
+ (
+ "directory-denied",
+ serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:*"]},{"Effect":"Deny","Action":["admin:MigrateTableCatalog"]}]}),
+ false,
+ ),
+ (
+ "directory-migrator",
+ serde_json::json!({"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:MigrateTableCatalog"]}]}),
+ true,
+ ),
+ ] {
+ iam.set_policy(name, Policy::parse_config(&serde_json::to_vec(&policy).unwrap()).unwrap())
+ .await
+ .unwrap();
+ iam.policy_db_set(access_key, UserType::Reg, false, name).await.unwrap();
+ for prefix in ["/iceberg/v1", "/_iceberg/v1"] {
+ let mut router = matchit::Router::new();
+ router
+ .insert(format!("{prefix}/{{warehouse}}/catalog/directory"), ())
+ .unwrap();
+ let uri = format!("{prefix}/warehouse/catalog/directory");
+ let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+ request.method = Method::DELETE;
+ request.uri = uri.parse().unwrap();
+ let matched = router.at(&uri).unwrap();
+ let result = DisableTableDirectoryHandler {}.call(request, matched.params).await;
+ if allowed {
+ assert_eq!(result.unwrap().output.0, StatusCode::OK);
+ } else {
+ assert_eq!(result.unwrap_err().code(), &S3ErrorCode::AccessDenied);
+ let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+ request.method = Method::POST;
+ request.uri = uri.parse().unwrap();
+ let matched = router.at(&uri).unwrap();
+ assert_eq!(
+ AdvanceTableDirectoryHandler {}
+ .call(request, matched.params)
+ .await
+ .unwrap_err()
+ .code(),
+ &S3ErrorCode::AccessDenied
+ );
+ }
+ if allowed {
+ for expected_phase in ["building", "ready"] {
+ let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+ request.method = Method::POST;
+ request.uri = uri.parse().unwrap();
+ let matched = router.at(&uri).unwrap();
+ let response = AdvanceTableDirectoryHandler {}.call(request, matched.params).await.unwrap();
+ assert_eq!(response.output.0, StatusCode::OK);
+ let bytes = http_body_util::BodyExt::collect(response.output.1).await.unwrap().to_bytes();
+ let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(value["phase"], expected_phase);
+ }
+ let mut request = table_catalog_handler_request(context.clone(), access_key, secret_key);
+ request.method = Method::DELETE;
+ request.uri = uri.parse().unwrap();
+ let matched = router.at(&uri).unwrap();
+ assert_eq!(DisableTableDirectoryHandler {}.call(request, matched.params).await.unwrap().output.0, StatusCode::OK);
+ }
+ let object_catalog = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone());
+ assert!(
+ object_catalog
+ .table_directory_status("warehouse")
+ .await
+ .unwrap()
+ .phase
+ .is_none(),
+ "denied requests and idempotent rollback must not activate a directory"
+ );
+ }
+ }
+ }).await;
+}
diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs
index 13a555a1c..76a639c4d 100644
--- a/rustfs/src/admin/route_policy.rs
+++ b/rustfs/src/admin/route_policy.rs
@@ -1132,6 +1132,24 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_BUCKET,
RouteRiskLevel::Sensitive,
),
+ admin(
+ HttpMethod::Get,
+ "/iceberg/v1/{warehouse}/catalog/directory",
+ GET_TABLE_CATALOG,
+ RouteRiskLevel::Sensitive,
+ ),
+ admin(
+ HttpMethod::Post,
+ "/iceberg/v1/{warehouse}/catalog/directory",
+ MIGRATE_TABLE_CATALOG,
+ RouteRiskLevel::High,
+ ),
+ admin(
+ HttpMethod::Delete,
+ "/iceberg/v1/{warehouse}/catalog/directory",
+ MIGRATE_TABLE_CATALOG,
+ RouteRiskLevel::High,
+ ),
admin(
HttpMethod::Get,
"/iceberg/v1/{warehouse}/catalog/migration",
@@ -1440,6 +1458,24 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_BUCKET,
RouteRiskLevel::Sensitive,
),
+ admin(
+ HttpMethod::Get,
+ "/_iceberg/v1/{warehouse}/catalog/directory",
+ GET_TABLE_CATALOG,
+ RouteRiskLevel::Sensitive,
+ ),
+ admin(
+ HttpMethod::Post,
+ "/_iceberg/v1/{warehouse}/catalog/directory",
+ MIGRATE_TABLE_CATALOG,
+ RouteRiskLevel::High,
+ ),
+ admin(
+ HttpMethod::Delete,
+ "/_iceberg/v1/{warehouse}/catalog/directory",
+ MIGRATE_TABLE_CATALOG,
+ RouteRiskLevel::High,
+ ),
admin(
HttpMethod::Get,
"/_iceberg/v1/{warehouse}/catalog/migration",
@@ -2125,7 +2161,7 @@ mod tests {
let table_specs = ADMIN_ROUTE_POLICY_SPECS
.iter()
.filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
- assert_eq!(table_specs.count(), 106);
+ assert_eq!(table_specs.count(), 112);
assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Delete, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Delete, "/_iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
@@ -2304,6 +2340,12 @@ mod tests {
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/quarantine",
RUN_TABLE_MAINTENANCE,
);
+ assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/catalog/directory", GET_TABLE_CATALOG);
+ assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+ assert_action(HttpMethod::Delete, "/iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+ assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/catalog/directory", GET_TABLE_CATALOG);
+ assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
+ assert_action(HttpMethod::Delete, "/_iceberg/v1/{warehouse}/catalog/directory", MIGRATE_TABLE_CATALOG);
assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/catalog/migration", GET_TABLE_CATALOG);
assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/catalog/migration", MIGRATE_TABLE_CATALOG);
diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs
index bb77f0c1a..ebb098a56 100644
--- a/rustfs/src/storage_api.rs
+++ b/rustfs/src/storage_api.rs
@@ -483,6 +483,7 @@ pub(crate) mod workload {
pub(crate) mod table {
pub(crate) use crate::storage::storage_api::ecstore_bucket::utils::check_valid_bucket_name_strict;
pub(crate) use crate::storage::storage_api::ecstore_object::{WriteCommitGuard, WriteCompletion};
+ pub(crate) use crate::storage::storage_api::request_context_consumer::spawn_traced_join;
pub(crate) mod contract {
pub(crate) mod http {
pub(crate) use super::super::super::storage_contracts::HTTPPreconditions;
@@ -518,6 +519,13 @@ pub(crate) mod table {
StorageObjectInfo, StorageObjectOptions, StorageObjectToDelete, StoragePutObjReader, get_bucket_metadata,
get_lock_acquire_timeout, table_catalog_path_hash,
};
+
+ #[cfg(test)]
+ pub(crate) mod test_support {
+ pub(crate) use crate::storage::storage_api::contract::bucket::{BucketOperations, MakeBucketOptions};
+ pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::init_bucket_metadata_sys;
+ pub(crate) use crate::storage::storage_api::ecstore_bucket::metadata_sys::test_support::isolated_store_over_temp_disks;
+ }
}
pub(crate) mod on_demand_migration {
diff --git a/rustfs/src/table_catalog/iceberg/manifest.rs b/rustfs/src/table_catalog/iceberg/manifest.rs
index a88eebfb8..0754e45ae 100644
--- a/rustfs/src/table_catalog/iceberg/manifest.rs
+++ b/rustfs/src/table_catalog/iceberg/manifest.rs
@@ -17,6 +17,7 @@ use std::io::Read;
use super::super::*;
const AVRO_ZSTANDARD_MAX_WINDOW_LOG: u32 = 27;
+const AVRO_MANIFEST_MAX_SCHEMA_DEPTH: usize = 64;
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct ManifestDataFileReference {
@@ -182,6 +183,7 @@ pub(crate) fn decode_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<
let decoded_size = validate_avro_container(data)?;
let reader = apache_avro::Reader::new(data)
.map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list Avro: {err}")))?;
+ validate_manifest_writer_schema(reader.writer_schema())?;
let format_version =
avro_record_format_version(reader.writer_schema(), &["sequence_number", "min_sequence_number"], "manifest list")?;
if format_version == 2 {
@@ -276,6 +278,7 @@ pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult TableCatalogStoreResult<()> {
+ use apache_avro::schema::{Name, NamesRef, ResolvedSchema};
+ use std::collections::{HashMap, HashSet};
+
+ fn depth(
+ schema: &apache_avro::Schema,
+ names: &NamesRef<'_>,
+ active: &mut HashSet,
+ heights: &mut HashMap,
+ nesting: usize,
+ ) -> TableCatalogStoreResult {
+ use apache_avro::Schema;
+ if nesting > AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+ return Err(TableCatalogStoreError::Invalid(
+ "manifest Avro schema exceeds the maximum nesting depth".to_string(),
+ ));
+ }
+ if let Schema::Ref { name } = schema {
+ let target = names.get(name).ok_or_else(|| {
+ TableCatalogStoreError::Invalid("manifest Avro schema has an unresolved type reference".to_string())
+ })?;
+ return depth(target, names, active, heights, nesting);
+ }
+ if let Some(name) = schema.name() {
+ if let Some(height) = heights.get(name) {
+ if nesting + height - 1 > AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+ return Err(TableCatalogStoreError::Invalid(
+ "manifest Avro schema exceeds the maximum nesting depth".to_string(),
+ ));
+ }
+ return Ok(*height);
+ }
+ if !active.insert(name.clone()) {
+ return Err(TableCatalogStoreError::Invalid(
+ "manifest Avro schema contains a recursive type".to_string(),
+ ));
+ }
+ }
+ let mut height = 1;
+ let mut visit = |child| -> TableCatalogStoreResult<()> {
+ height = height.max(1 + depth(child, names, active, heights, nesting + 1)?);
+ Ok(())
+ };
+ match schema {
+ Schema::Record(record) => {
+ for field in &record.fields {
+ visit(&field.schema)?;
+ }
+ }
+ Schema::Array(array) => visit(&array.items)?,
+ Schema::Map(map) => visit(&map.types)?,
+ Schema::Union(union) => {
+ for variant in union.variants() {
+ visit(variant)?;
+ }
+ }
+ _ => {}
+ }
+ if let Some(name) = schema.name() {
+ active.remove(name);
+ heights.insert(name.clone(), height);
+ }
+ Ok(height)
+ }
+
+ let resolved = ResolvedSchema::try_from(schema)
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("invalid manifest Avro schema: {err}")))?;
+ depth(schema, resolved.get_names(), &mut HashSet::new(), &mut HashMap::new(), 1)?;
+ Ok(())
+}
+
fn validate_avro_container(data: &[u8]) -> TableCatalogStoreResult {
if !data.starts_with(b"Obj\x01") {
return Err(TableCatalogStoreError::Invalid("Avro container has invalid header magic".to_string()));
@@ -665,6 +741,96 @@ mod tests {
use super::*;
use std::assert_matches;
+ fn schema_container(schema: &serde_json::Value, record: &[u8]) -> Vec {
+ fn long(bytes: &mut Vec, value: usize) {
+ let mut encoded = u64::try_from(value).expect("fixture length") << 1;
+ while encoded >= 128 {
+ bytes.push((encoded as u8 & 0x7f) | 0x80);
+ encoded >>= 7;
+ }
+ bytes.push(encoded as u8);
+ }
+ let schema = serde_json::to_vec(schema).expect("schema JSON");
+ let mut bytes = b"Obj\x01".to_vec();
+ long(&mut bytes, 1);
+ long(&mut bytes, b"avro.schema".len());
+ bytes.extend_from_slice(b"avro.schema");
+ long(&mut bytes, schema.len());
+ bytes.extend_from_slice(&schema);
+ long(&mut bytes, 0);
+ bytes.extend_from_slice(&[0; 16]);
+ long(&mut bytes, 1);
+ long(&mut bytes, record.len());
+ bytes.extend_from_slice(record);
+ bytes.extend_from_slice(&[0; 16]);
+ bytes
+ }
+
+ #[test]
+ fn manifest_decoders_reject_recursive_writer_schemas_before_decoding_values() {
+ for child in [
+ serde_json::json!(["null", "example.manifest_file"]),
+ serde_json::json!({"type":"array", "items":"example.manifest_file"}),
+ serde_json::json!({"type":"map", "values":"example.manifest_file"}),
+ serde_json::json!({"type":"record", "name":"child", "fields":[{"name":"parent", "type":"example.manifest_file"}]}),
+ ] {
+ let schema = serde_json::json!({
+ "type":"record", "name":"manifest_file", "namespace":"example",
+ "fields":[{"name":"manifest_path","type":"string"}, {"name":"next", "type":child}]
+ });
+ // A valid self-referential union datum can drive thousands of decoder frames.
+ let mut record = [0, 2].repeat(10_000);
+ record.extend_from_slice(&[0, 0]);
+ let data = schema_container(&schema, &record);
+ assert!(
+ matches!(decode_manifest_list_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("recursive type"))
+ );
+ assert!(
+ matches!(decode_manifest_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("recursive type"))
+ );
+ }
+ }
+
+ #[test]
+ fn manifest_decoders_bound_resolved_schema_depth() {
+ let mut fields = vec![serde_json::json!({"name":"manifest_path", "type":"string"})];
+ for index in 0..AVRO_MANIFEST_MAX_SCHEMA_DEPTH {
+ let children = if index == 0 {
+ vec![]
+ } else {
+ vec![serde_json::json!({"name":"previous", "type":format!("node{}", index - 1)})]
+ };
+ fields.push(serde_json::json!({"name":format!("field{index}"), "type":{"type":"record", "name":format!("node{index}"), "fields":children}}));
+ }
+ let data = schema_container(&serde_json::json!({"type":"record", "name":"manifest_file", "fields":fields}), &[0]);
+ assert!(
+ matches!(decode_manifest_list_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("maximum nesting depth"))
+ );
+ assert!(
+ matches!(decode_manifest_avro(&data), Err(TableCatalogStoreError::Invalid(message)) if message.contains("maximum nesting depth"))
+ );
+ }
+
+ #[test]
+ fn manifest_decoder_accepts_reused_nonrecursive_named_types() {
+ let schema = serde_json::json!({
+ "type":"record", "name":"manifest_file", "namespace":"example",
+ "fields":[
+ {"name":"manifest_path", "type":"string"},
+ {"name":"first", "type":{"type":"record", "name":"detail", "fields":[{"name":"value", "type":"string"}]}},
+ {"name":"second", "type":"detail"}
+ ]
+ });
+ let path = "manifest.avro";
+ let mut record = vec![u8::try_from(path.len() * 2).expect("short path")];
+ record.extend_from_slice(path.as_bytes());
+ record.extend_from_slice(&[0, 0]);
+ let decoded =
+ decode_manifest_list_avro(&schema_container(&schema, &record)).expect("reused named types are not recursive");
+ assert_eq!(decoded.references.len(), 1);
+ assert_eq!(decoded.references[0].manifest_path, path);
+ }
+
#[test]
fn rejects_v2_manifest_lists_without_required_count_fields() {
let schema = apache_avro::Schema::parse_str(
diff --git a/rustfs/src/table_catalog/maintenance/compaction.rs b/rustfs/src/table_catalog/maintenance/compaction.rs
new file mode 100644
index 000000000..23079908b
--- /dev/null
+++ b/rustfs/src/table_catalog/maintenance/compaction.rs
@@ -0,0 +1,924 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::super::*;
+use datafusion::arrow::array::{
+ Array, BinaryArray, BinaryViewArray, FixedSizeListArray, LargeBinaryArray, LargeListArray, LargeListViewArray,
+ LargeStringArray, ListArray, ListViewArray, MapArray, StringArray, StringViewArray, StructArray, downcast_dictionary_array,
+};
+use datafusion::arrow::datatypes::{DataType, Field, Schema};
+use datafusion::parquet::arrow::{
+ add_encoded_arrow_schema_to_metadata, arrow_reader::ArrowReaderOptions, arrow_writer::ArrowWriterOptions,
+};
+use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties};
+use std::io::Write;
+use std::sync::OnceLock;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore};
+
+const COMPACTION_WRITE_BYTES: usize = 8 * 1024 * 1024;
+const COMPACTION_BATCH_ROWS: usize = 256;
+const COMPACTION_ROW_GROUP_ROWS: usize = 65_536;
+const COMPACTION_MAX_ROW_GROUPS: usize = 256;
+const COMPACTION_MAX_COLUMN_CHUNKS: usize = 16_384;
+const COMPACTION_SCHEMA_DEPTH: usize = 64;
+const COMPACTION_SCHEMA_FIELDS: usize = 1024;
+
+pub(crate) fn try_admit_compaction() -> TableCatalogStoreResult> {
+ static ADMISSION: OnceLock> = OnceLock::new();
+ ADMISSION
+ .get_or_init(|| Arc::new(Semaphore::new(1)))
+ .clone()
+ .try_acquire_owned()
+ .map(Arc::new)
+ .map_err(|_| TableCatalogStoreError::Unavailable("table compaction capacity is full".to_owned()))
+}
+
+fn budget_error() -> TableCatalogStoreError {
+ TableCatalogStoreError::Invalid("table compaction exceeds the server memory or expanded-value budget".to_owned())
+}
+
+fn add_budget(total: &mut usize, amount: usize, limit: usize) -> TableCatalogStoreResult<()> {
+ *total = total
+ .checked_add(amount)
+ .filter(|value| *value <= limit)
+ .ok_or_else(budget_error)?;
+ Ok(())
+}
+
+/// Sizes read from one Parquet page header (Thrift compact encoding).
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+struct PageHeaderSizes {
+ uncompressed: usize,
+ compressed: usize,
+}
+
+const THRIFT_COMPACT_STOP: u8 = 0;
+const THRIFT_COMPACT_BOOL_TRUE: u8 = 1;
+const THRIFT_COMPACT_BOOL_FALSE: u8 = 2;
+const THRIFT_COMPACT_BYTE: u8 = 3;
+const THRIFT_COMPACT_I16: u8 = 4;
+const THRIFT_COMPACT_I32: u8 = 5;
+const THRIFT_COMPACT_I64: u8 = 6;
+const THRIFT_COMPACT_DOUBLE: u8 = 7;
+const THRIFT_COMPACT_BINARY: u8 = 8;
+const THRIFT_COMPACT_LIST: u8 = 9;
+const THRIFT_COMPACT_SET: u8 = 10;
+const THRIFT_COMPACT_MAP: u8 = 11;
+const THRIFT_COMPACT_STRUCT: u8 = 12;
+const THRIFT_MAX_NESTING: usize = 16;
+const PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE: i16 = 2;
+const PAGE_HEADER_FIELD_COMPRESSED_SIZE: i16 = 3;
+
+/// Minimal Thrift compact-protocol cursor: enough to read a `PageHeader`'s
+/// size fields and skip everything else so the next header can be located.
+struct ThriftCompactCursor<'a> {
+ bytes: &'a [u8],
+ pos: usize,
+}
+
+impl ThriftCompactCursor<'_> {
+ fn error() -> TableCatalogStoreError {
+ TableCatalogStoreError::Invalid("compaction parquet page header is malformed".to_owned())
+ }
+
+ fn byte(&mut self) -> TableCatalogStoreResult {
+ let byte = *self.bytes.get(self.pos).ok_or_else(Self::error)?;
+ self.pos += 1;
+ Ok(byte)
+ }
+
+ fn skip_bytes(&mut self, count: usize) -> TableCatalogStoreResult<()> {
+ self.pos = self
+ .pos
+ .checked_add(count)
+ .filter(|end| *end <= self.bytes.len())
+ .ok_or_else(Self::error)?;
+ Ok(())
+ }
+
+ fn varint(&mut self) -> TableCatalogStoreResult {
+ let mut value = 0_u64;
+ for shift in (0..64).step_by(7) {
+ let byte = self.byte()?;
+ value |= u64::from(byte & 0x7f) << shift;
+ if byte & 0x80 == 0 {
+ return Ok(value);
+ }
+ }
+ Err(Self::error())
+ }
+
+ fn zigzag(&mut self) -> TableCatalogStoreResult {
+ let raw = self.varint()?;
+ Ok(((raw >> 1) as i64) ^ -((raw & 1) as i64))
+ }
+
+ fn skip_value(&mut self, kind: u8, depth: usize) -> TableCatalogStoreResult<()> {
+ if depth > THRIFT_MAX_NESTING {
+ return Err(Self::error());
+ }
+ match kind {
+ THRIFT_COMPACT_BOOL_TRUE | THRIFT_COMPACT_BOOL_FALSE => Ok(()),
+ THRIFT_COMPACT_BYTE => self.skip_bytes(1),
+ THRIFT_COMPACT_I16 | THRIFT_COMPACT_I32 | THRIFT_COMPACT_I64 => self.varint().map(drop),
+ THRIFT_COMPACT_DOUBLE => self.skip_bytes(8),
+ THRIFT_COMPACT_BINARY => {
+ let len = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
+ self.skip_bytes(len)
+ }
+ THRIFT_COMPACT_LIST | THRIFT_COMPACT_SET => {
+ let header = self.byte()?;
+ let element = header & 0x0f;
+ let count = match header >> 4 {
+ 15 => usize::try_from(self.varint()?).map_err(|_| Self::error())?,
+ short => usize::from(short),
+ };
+ for _ in 0..count {
+ self.skip_value(element, depth + 1)?;
+ }
+ Ok(())
+ }
+ THRIFT_COMPACT_MAP => {
+ let count = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
+ if count == 0 {
+ return Ok(());
+ }
+ let kinds = self.byte()?;
+ for _ in 0..count {
+ self.skip_value(kinds >> 4, depth + 1)?;
+ self.skip_value(kinds & 0x0f, depth + 1)?;
+ }
+ Ok(())
+ }
+ THRIFT_COMPACT_STRUCT => self.skip_struct(depth + 1),
+ _ => Err(Self::error()),
+ }
+ }
+
+ fn skip_struct(&mut self, depth: usize) -> TableCatalogStoreResult<()> {
+ let mut field_id = 0_i16;
+ loop {
+ let header = self.byte()?;
+ if header == THRIFT_COMPACT_STOP {
+ return Ok(());
+ }
+ let kind = header & 0x0f;
+ field_id = match header >> 4 {
+ 0 => i16::try_from(self.zigzag()?).map_err(|_| Self::error())?,
+ delta => field_id.checked_add(i16::from(delta)).ok_or_else(Self::error)?,
+ };
+ self.skip_value(kind, depth)?;
+ }
+ }
+
+ /// Read a `PageHeader` struct, returning its sizes and leaving the cursor at
+ /// the first page byte.
+ fn page_header(&mut self) -> TableCatalogStoreResult {
+ let mut sizes = PageHeaderSizes::default();
+ let mut field_id = 0_i16;
+ loop {
+ let header = self.byte()?;
+ if header == THRIFT_COMPACT_STOP {
+ return Ok(sizes);
+ }
+ let kind = header & 0x0f;
+ field_id = match header >> 4 {
+ 0 => i16::try_from(self.zigzag()?).map_err(|_| Self::error())?,
+ delta => field_id.checked_add(i16::from(delta)).ok_or_else(Self::error)?,
+ };
+ match (field_id, kind) {
+ (PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE, THRIFT_COMPACT_I32) => {
+ sizes.uncompressed = usize::try_from(self.zigzag()?).map_err(|_| Self::error())?;
+ }
+ (PAGE_HEADER_FIELD_COMPRESSED_SIZE, THRIFT_COMPACT_I32) => {
+ sizes.compressed = usize::try_from(self.zigzag()?).map_err(|_| Self::error())?;
+ }
+ _ => self.skip_value(kind, 1)?,
+ }
+ }
+ }
+}
+
+/// Sum the page headers of one column chunk (`[start, end)` in the file) and
+/// hold every page and the chunk total to `limit`. Pages are never decompressed.
+fn validate_page_headers_in_range(chunk: &[u8], chunk_declared: usize, limit: usize) -> TableCatalogStoreResult {
+ let mut cursor = ThriftCompactCursor { bytes: chunk, pos: 0 };
+ let mut total = 0_usize;
+ while cursor.pos < chunk.len() {
+ let sizes = cursor.page_header()?;
+ add_budget(&mut total, sizes.uncompressed, limit)?;
+ if total > chunk_declared.max(limit.min(chunk_declared.saturating_mul(2))) {
+ return Err(TableCatalogStoreError::Invalid(
+ "compaction parquet page headers exceed the footer's declared chunk size".to_owned(),
+ ));
+ }
+ cursor.skip_bytes(sizes.compressed)?;
+ }
+ Ok(total)
+}
+
+fn validate_column_chunk_page_headers(
+ data: &Bytes,
+ column: &datafusion::parquet::file::metadata::ColumnChunkMetaData,
+ chunk_declared: usize,
+ limit: usize,
+ location: &str,
+) -> TableCatalogStoreResult<()> {
+ let start = column
+ .dictionary_page_offset()
+ .map_or(column.data_page_offset(), |dictionary| dictionary.min(column.data_page_offset()));
+ let start = usize::try_from(start).map_err(|_| budget_error())?;
+ let length = usize::try_from(column.compressed_size()).map_err(|_| budget_error())?;
+ let end = start.checked_add(length).filter(|end| *end <= data.len()).ok_or_else(|| {
+ TableCatalogStoreError::Invalid(format!("compaction parquet {location} column chunk lies outside the file"))
+ })?;
+ validate_page_headers_in_range(&data[start..end], chunk_declared, limit).map_err(|err| match err {
+ TableCatalogStoreError::Invalid(message) => TableCatalogStoreError::Invalid(format!("{location}: {message}")),
+ other => other,
+ })?;
+ Ok(())
+}
+
+fn view_field(field: &Field, depth: usize, fields: &mut usize) -> TableCatalogStoreResult> {
+ add_budget(fields, 1, COMPACTION_SCHEMA_FIELDS)?;
+ Ok(Arc::new(field.clone().with_data_type(view_type(field.data_type(), depth, fields)?)))
+}
+
+fn view_type(data_type: &DataType, depth: usize, fields: &mut usize) -> TableCatalogStoreResult {
+ if depth > COMPACTION_SCHEMA_DEPTH {
+ return Err(budget_error());
+ }
+ Ok(match data_type {
+ DataType::Utf8 | DataType::LargeUtf8 => DataType::Utf8View,
+ DataType::Binary | DataType::LargeBinary => DataType::BinaryView,
+ DataType::List(field) => DataType::List(view_field(field, depth + 1, fields)?),
+ DataType::LargeList(field) => DataType::LargeList(view_field(field, depth + 1, fields)?),
+ DataType::ListView(field) => DataType::ListView(view_field(field, depth + 1, fields)?),
+ DataType::LargeListView(field) => DataType::LargeListView(view_field(field, depth + 1, fields)?),
+ DataType::FixedSizeList(field, size) => DataType::FixedSizeList(view_field(field, depth + 1, fields)?, *size),
+ DataType::Map(field, sorted) => DataType::Map(view_field(field, depth + 1, fields)?, *sorted),
+ DataType::Struct(children) => DataType::Struct(
+ children
+ .iter()
+ .map(|field| view_field(field, depth + 1, fields))
+ .collect::>>()?
+ .into(),
+ ),
+ DataType::Dictionary(_, value) => {
+ // The pinned reader does not support Dictionary. Keep
+ // its already shared values, while still checking nested schema limits.
+ let _ = view_type(value, depth + 1, fields)?;
+ data_type.clone()
+ }
+ other => other.clone(),
+ })
+}
+
+fn view_schema(schema: &Schema) -> TableCatalogStoreResult {
+ let mut fields = 0;
+ let mapped = schema
+ .fields()
+ .iter()
+ .map(|field| view_field(field, 0, &mut fields))
+ .collect::>>()?;
+ Ok(Arc::new(Schema::new_with_metadata(mapped, schema.metadata().clone())))
+}
+
+/// Charge the values the writer may expand, not the small view/dictionary
+/// buffers. Even a null or empty nested cell consumes structural capacity.
+fn charge_value(array: &dyn Array, row: usize, depth: usize, total: &mut usize, limit: usize) -> TableCatalogStoreResult<()> {
+ if depth > COMPACTION_SCHEMA_DEPTH {
+ return Err(budget_error());
+ }
+ add_budget(total, 8, limit)?;
+ if array.is_null(row) {
+ return Ok(());
+ }
+ macro_rules! scalar_bytes {
+ ($ty:ty) => {
+ add_budget(
+ total,
+ array
+ .as_any()
+ .downcast_ref::<$ty>()
+ .ok_or_else(budget_error)?
+ .value(row)
+ .len(),
+ limit,
+ )
+ };
+ }
+ macro_rules! nested_values {
+ ($ty:ty) => {{
+ let value = array.as_any().downcast_ref::<$ty>().ok_or_else(budget_error)?.value(row);
+ for index in 0..value.len() {
+ charge_value(value.as_ref(), index, depth + 1, total, limit)?;
+ }
+ Ok(())
+ }};
+ }
+ match array.data_type() {
+ DataType::Utf8 => scalar_bytes!(StringArray),
+ DataType::LargeUtf8 => scalar_bytes!(LargeStringArray),
+ DataType::Utf8View => scalar_bytes!(StringViewArray),
+ DataType::Binary => scalar_bytes!(BinaryArray),
+ DataType::LargeBinary => scalar_bytes!(LargeBinaryArray),
+ DataType::BinaryView => scalar_bytes!(BinaryViewArray),
+ DataType::Boolean | DataType::Null => Ok(()),
+ DataType::FixedSizeBinary(size) => add_budget(total, usize::try_from(*size).map_err(|_| budget_error())?, limit),
+ DataType::List(_) => nested_values!(ListArray),
+ DataType::LargeList(_) => nested_values!(LargeListArray),
+ DataType::ListView(_) => nested_values!(ListViewArray),
+ DataType::LargeListView(_) => nested_values!(LargeListViewArray),
+ DataType::FixedSizeList(_, _) => nested_values!(FixedSizeListArray),
+ DataType::Map(_, _) => {
+ let value = array.as_any().downcast_ref::().ok_or_else(budget_error)?.value(row);
+ for index in 0..value.len() {
+ charge_value(&value, index, depth + 1, total, limit)?;
+ }
+ Ok(())
+ }
+ DataType::Struct(_) => {
+ for column in array
+ .as_any()
+ .downcast_ref::()
+ .ok_or_else(budget_error)?
+ .columns()
+ {
+ charge_value(column.as_ref(), row, depth + 1, total, limit)?;
+ }
+ Ok(())
+ }
+ DataType::Dictionary(_, _) => downcast_dictionary_array!(array => {
+ if let Some(key) = array.key(row) { charge_value(array.values().as_ref(), key, depth + 1, total, limit)?; }
+ Ok(())
+ }, _ => Err(budget_error())),
+ data_type => add_budget(total, data_type.primitive_width().ok_or_else(budget_error)?, limit),
+ }
+}
+
+struct BoundedOutput {
+ bytes: Vec,
+ limit: usize,
+}
+impl Write for BoundedOutput {
+ fn write(&mut self, bytes: &[u8]) -> std::io::Result {
+ let Some(required) = self.bytes.len().checked_add(bytes.len()).filter(|size| *size <= self.limit) else {
+ return Err(std::io::Error::other("compaction output exceeds the server size limit"));
+ };
+ if required > self.bytes.capacity() {
+ // Parquet writes successive pages and buffered footer fragments.
+ // Amortize their copies without allocating past the output budget.
+ let capacity = required.max(self.bytes.capacity().saturating_mul(2)).min(self.limit);
+ self.bytes
+ .try_reserve_exact(capacity - self.bytes.len())
+ .map_err(std::io::Error::other)?;
+ }
+ self.bytes.extend_from_slice(bytes);
+ Ok(bytes.len())
+ }
+ fn flush(&mut self) -> std::io::Result<()> {
+ Ok(())
+ }
+}
+
+pub(crate) fn compact_parquet_data_files(input_files: Vec<(String, Vec)>) -> TableCatalogStoreResult {
+ compact_parquet_with_limits(input_files, TABLE_COMPACTION_MAX_DECODED_BYTES, COMPACTION_WRITE_BYTES)
+}
+
+fn compact_parquet_with_limits(
+ input_files: Vec<(String, Vec)>,
+ limit: usize,
+ write_limit: usize,
+) -> TableCatalogStoreResult {
+ compact_parquet_with_budgets(input_files, limit, write_limit, COMPACTION_MAX_ROW_GROUPS, COMPACTION_ROW_GROUP_ROWS)
+}
+
+fn compact_parquet_with_budgets(
+ input_files: Vec<(String, Vec)>,
+ limit: usize,
+ write_limit: usize,
+ group_limit: usize,
+ row_group_rows: usize,
+) -> TableCatalogStoreResult {
+ let mut schema: Option = None;
+ let mut declared = 0;
+ let mut stored = 0;
+ let mut files = Vec::new();
+ let mut leaf_columns = 0;
+ // Validate every footer before decoding any page or creating output.
+ for (location, data) in input_files {
+ add_budget(&mut stored, data.len(), TABLE_COMPACTION_MAX_STORED_BYTES)?;
+ let data = Bytes::from(data);
+ let builder = ParquetRecordBatchReaderBuilder::try_new(data.clone())
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read compaction parquet {location}: {err}")))?;
+ let file_schema = builder.schema().clone();
+ leaf_columns = leaf_columns.max(builder.metadata().file_metadata().schema_descr().num_columns());
+ match &schema {
+ Some(expected) if expected.as_ref() != file_schema.as_ref() => {
+ return Err(TableCatalogStoreError::Invalid("compaction input parquet schemas must match".to_owned()));
+ }
+ None => schema = Some(file_schema),
+ _ => {}
+ }
+ for group in builder.metadata().row_groups() {
+ let mut columns = 0;
+ for column in group.columns() {
+ let chunk_declared = usize::try_from(column.uncompressed_size()).map_err(|_| budget_error())?;
+ add_budget(&mut columns, chunk_declared, limit)?;
+ // The footer is one claim; the page headers the decoder actually
+ // sizes its buffers from are another. Walk them without
+ // decompressing and hold both to the budget.
+ validate_column_chunk_page_headers(&data, column, chunk_declared, limit, &location)?;
+ }
+ let group_bytes = usize::try_from(group.total_byte_size()).map_err(|_| budget_error())?;
+ add_budget(&mut declared, group_bytes.max(columns), limit)?;
+ }
+ files.push((location, data));
+ }
+ let schema = schema
+ .ok_or_else(|| TableCatalogStoreError::Invalid("compaction requires at least one parquet input file".to_owned()))?;
+ let read_schema = view_schema(&schema)?;
+ let group_limit = group_limit.min(COMPACTION_MAX_COLUMN_CHUNKS / leaf_columns.max(1));
+ // These bounds include metadata retained until the footer is serialized.
+ // Page/offset indexes would add a separate, page-count-dependent inventory.
+ let mut properties = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(row_group_rows))
+ .set_statistics_enabled(EnabledStatistics::Chunk)
+ .set_statistics_truncate_length(Some(64))
+ .set_offset_index_disabled(true)
+ .build();
+ // The writer requires nested array types to match its input schema exactly.
+ // Views share the same Parquet physical/logical types; retain the original
+ // Arrow schema in the footer so readers recover its types and field metadata.
+ // Do not cast sliced lists/maps: Arrow casts their entire child buffers.
+ add_encoded_arrow_schema_to_metadata(&schema, &mut properties);
+ let options = ArrowWriterOptions::new()
+ .with_properties(properties)
+ .with_skip_arrow_metadata(true);
+ let mut writer = ArrowWriter::try_new_with_options(
+ BoundedOutput {
+ bytes: Vec::new(),
+ limit,
+ },
+ read_schema.clone(),
+ options,
+ )
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to build compaction writer: {err}")))?;
+ let mut record_count = 0_u64;
+ let mut decoded = 0;
+ for (location, data) in files {
+ let options = ArrowReaderOptions::new().with_schema(read_schema.clone());
+ let reader = ParquetRecordBatchReaderBuilder::try_new_with_options(data, options)
+ .and_then(|builder| builder.with_batch_size(COMPACTION_BATCH_ROWS).build())
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to build compaction reader {location}: {err}")))?;
+ for batch in reader {
+ let batch = batch
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to decode compaction batch {location}: {err}")))?;
+ add_budget(&mut decoded, batch.get_array_memory_size(), limit)?;
+ let mut start = 0;
+ while start < batch.num_rows() {
+ let mut end = start;
+ let mut expanded = 0_usize;
+ let remaining_rows = row_group_rows.saturating_sub(writer.in_progress_rows());
+ while end < batch.num_rows() && end - start < remaining_rows {
+ let mut row_bytes = 0;
+ for column in batch.columns() {
+ charge_value(column.as_ref(), end, 0, &mut row_bytes, write_limit)?;
+ }
+ if expanded.checked_add(row_bytes).is_none_or(|size| size > write_limit) {
+ break;
+ }
+ expanded += row_bytes;
+ end += 1;
+ }
+ if end == start {
+ return Err(budget_error());
+ }
+ if writer
+ .memory_size()
+ .checked_add(expanded)
+ .is_none_or(|size| size > write_limit)
+ {
+ writer.flush().map_err(|err| {
+ TableCatalogStoreError::Invalid(format!("failed to flush bounded compaction output: {err}"))
+ })?;
+ }
+ if writer.flushed_row_groups().len() >= group_limit {
+ return Err(TableCatalogStoreError::Invalid(
+ "compaction output row-group metadata exceeds the server limit".to_owned(),
+ ));
+ }
+ writer.write(&batch.slice(start, end - start)).map_err(|err| {
+ TableCatalogStoreError::Invalid(format!("failed to write bounded compaction output: {err}"))
+ })?;
+ record_count = record_count.checked_add((end - start) as u64).ok_or_else(budget_error)?;
+ start = end;
+ }
+ }
+ }
+ let output = writer
+ .into_inner()
+ .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to finalize bounded compaction output: {err}")))?;
+ Ok(CompactedParquetFile {
+ data: output.bytes,
+ record_count,
+ })
+}
+
+pub(crate) async fn run_compaction_worker(
+ input_files: Vec<(String, Vec)>,
+ admission: Arc,
+) -> TableCatalogStoreResult {
+ #[cfg(test)]
+ let barrier = COMPACTION_WORKER_BARRIER.try_with(Arc::clone).ok();
+ tokio::task::spawn_blocking(move || {
+ let _admission = admission;
+ #[cfg(test)]
+ if let Some(barrier) = barrier {
+ barrier.entered.notify_one();
+ let mut released = barrier.released.lock().unwrap();
+ while !*released {
+ released = barrier.release.wait(released).unwrap();
+ }
+ }
+ compact_parquet_data_files(input_files)
+ })
+ .await
+ .map_err(|_| TableCatalogStoreError::Internal("compaction decoder worker failed".to_owned()))?
+}
+
+pub(crate) async fn run_compaction_output(operation: F, admission: Arc) -> TableCatalogStoreResult<()>
+where
+ F: std::future::Future