gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(table-catalog): add maintenance worker runtime (#3480)

Browse Source
This commit is contained in:
Henry Guo
2026-06-15 18:42:29 +08:00
committed by GitHub
parent f782f94f6a
commit 7ce9838027
6 changed files with 816 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog.rs
+136 -17
View File
@@ -72,6 +72,7 @@ const S3_SESSION_TOKEN_CONFIG_KEY: &str = "s3.session-token";
const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces";
const TABLE_CATALOG_TABLE_RESOURCE_ROOT: &str = "tables";
const TABLE_CATALOG_ADMIN_OPERATION_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2);
const DEFAULT_TABLE_MAINTENANCE_WORKER_ID: &str = "rustfs-maintenance-worker";
const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
"GET /v1/{prefix}/namespaces",
"POST /v1/{prefix}/namespaces",
@@ -113,6 +114,8 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[
"GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config",
"PUT /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config",
"GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}",
"POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
"POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
"GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
"POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import",
"GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external",
@@ -149,6 +152,8 @@ static TABLE_METADATA_MAINTENANCE_HANDLER: RestTableMetadataMaintenanceHandler =
static GET_TABLE_MAINTENANCE_CONFIG_HANDLER: GetTableMaintenanceConfigHandler = GetTableMaintenanceConfigHandler {};
static PUT_TABLE_MAINTENANCE_CONFIG_HANDLER: PutTableMaintenanceConfigHandler = PutTableMaintenanceConfigHandler {};
static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTableMaintenanceJobHandler {};
static RUN_TABLE_MAINTENANCE_WORKER_HANDLER: RunTableMaintenanceWorkerHandler = RunTableMaintenanceWorkerHandler {};
static HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER: HeartbeatTableMaintenanceJobHandler = HeartbeatTableMaintenanceJobHandler {};
static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {};
static IMPORT_TABLE_CATALOG_HANDLER: ImportTableCatalogHandler = ImportTableCatalogHandler {};
static EXTERNAL_CATALOG_BRIDGE_HANDLER: ExternalCatalogBridgeHandler = ExternalCatalogBridgeHandler {};
@@ -238,6 +243,28 @@ struct TableMetadataMaintenanceRequest {
compaction: Option<crate::table_catalog::TableCompactionPlanningConfig>,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct TableMaintenanceWorkerRunRequest {
#[serde(default, rename = "worker-id")]
worker_id: Option<String>,
}
impl TableMaintenanceWorkerRunRequest {
fn worker_id(&self) -> &str {
self.worker_id.as_deref().unwrap_or(DEFAULT_TABLE_MAINTENANCE_WORKER_ID)
}
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct TableMaintenanceHeartbeatRequest {
#[serde(rename = "lease-id")]
lease_id: String,
#[serde(rename = "worker-id")]
worker_id: String,
}
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct UpdateTableMetadataLocationRequest {
@@ -687,6 +714,16 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/jobs/{{job}}").as_str(),
AdminOperation(&GET_TABLE_MAINTENANCE_JOB_HANDLER),
)?;
r.insert(
Method::POST,
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/worker/run").as_str(),
AdminOperation(&RUN_TABLE_MAINTENANCE_WORKER_HANDLER),
)?;
r.insert(
Method::POST,
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/jobs/{{job}}/heartbeat").as_str(),
AdminOperation(&HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER),
)?;
r.insert(
Method::GET,
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/export").as_str(),
@@ -3365,6 +3402,59 @@ impl Operation for GetTableMaintenanceJobHandler {
}
}
pub struct RunTableMaintenanceWorkerHandler {}
#[async_trait::async_trait]
impl Operation for RunTableMaintenanceWorkerHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let table = table_name_from_params(&params)?;
let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
let request = read_json_body::<TableMaintenanceWorkerRunRequest>(req.input).await?;
let store = table_catalog_store()?;
let response = store
.run_table_metadata_maintenance_worker_once(
&warehouse,
&namespace.public_name(),
&table,
request.worker_id().to_string(),
)
.await
.map_err(catalog_store_error)?;
build_json_response(StatusCode::OK, &response)
}
}
pub struct HeartbeatTableMaintenanceJobHandler {}
#[async_trait::async_trait]
impl Operation for HeartbeatTableMaintenanceJobHandler {
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let table = table_name_from_params(&params)?;
let job = job_id_from_params(&params)?;
let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?;
let request = read_json_body::<TableMaintenanceHeartbeatRequest>(req.input).await?;
let store = table_catalog_store()?;
let response = store
.heartbeat_table_metadata_maintenance_job(
&warehouse,
&namespace.public_name(),
&table,
&job,
&request.lease_id,
&request.worker_id,
)
.await
.map_err(catalog_store_error)?;
build_json_response(StatusCode::OK, &response)
}
}
pub struct ExportTableCatalogHandler {}
#[async_trait::async_trait]
@@ -3785,6 +3875,8 @@ mod tests {
assert_operation::<GetTableMaintenanceConfigHandler>();
assert_operation::<PutTableMaintenanceConfigHandler>();
assert_operation::<GetTableMaintenanceJobHandler>();
assert_operation::<RunTableMaintenanceWorkerHandler>();
assert_operation::<HeartbeatTableMaintenanceJobHandler>();
assert_operation::<ExportTableCatalogHandler>();
assert_operation::<ImportTableCatalogHandler>();
assert_operation::<ExternalCatalogBridgeHandler>();
@@ -3850,6 +3942,34 @@ mod tests {
assert_eq!(compaction.max_rewrite_bytes_per_job, 10_737_418_240);
}
#[test]
fn table_maintenance_worker_run_request_uses_stable_default_worker_id() {
let request: TableMaintenanceWorkerRunRequest =
serde_json::from_value(serde_json::json!({})).expect("worker run request should parse");
assert_eq!(request.worker_id(), "rustfs-maintenance-worker");
}
#[test]
fn table_maintenance_worker_run_request_accepts_worker_id() {
let request: TableMaintenanceWorkerRunRequest = serde_json::from_value(serde_json::json!({
"worker-id": "worker-a"
}))
.expect("worker run request should parse worker id");
assert_eq!(request.worker_id(), "worker-a");
}
#[test]
fn table_maintenance_heartbeat_request_requires_lease_id() {
let err = serde_json::from_value::<TableMaintenanceHeartbeatRequest>(serde_json::json!({
"worker-id": "worker-a"
}))
.expect_err("heartbeat request should require lease id");
assert!(err.to_string().contains("lease-id"));
}
#[tokio::test]
async fn table_bucket_response_reports_catalog_discovery_without_credentials() {
let store = TestTableCatalogStore::default();
@@ -4752,23 +4872,22 @@ mod tests {
.expect("maintenance config should persist");
assert_eq!(config.retain_recent_metadata_files, 2);
assert!(config.delete_enabled);
assert!(
store
.put_table_maintenance_config(
bucket,
"analytics",
"events",
crate::table_catalog::TableMaintenanceConfig {
version: crate::table_catalog::TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 2,
delete_enabled: true,
background_enabled: true,
..Default::default()
},
)
.await
.is_err()
);
let background_config = store
.put_table_maintenance_config(
bucket,
"analytics",
"events",
crate::table_catalog::TableMaintenanceConfig {
version: crate::table_catalog::TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 2,
delete_enabled: true,
background_enabled: true,
..Default::default()
},
)
.await
.expect("background maintenance config should persist");
assert!(background_config.background_enabled);
let dry_run = table_metadata_maintenance_response(
&store,
rustfs/src/admin/route_policy.rs
+35 -1
View File
@@ -772,6 +772,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_LIFECYCLE,
RouteRiskLevel::Sensitive,
),
admin(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
RUN_TABLE_MAINTENANCE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
RUN_TABLE_MAINTENANCE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -971,6 +983,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
GET_TABLE_LIFECYCLE,
RouteRiskLevel::Sensitive,
),
admin(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
RUN_TABLE_MAINTENANCE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
RUN_TABLE_MAINTENANCE,
RouteRiskLevel::High,
),
admin(
HttpMethod::Get,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -1190,7 +1214,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(), 68);
assert_eq!(table_specs.count(), 72);
assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET);
assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
@@ -1300,6 +1324,16 @@ mod tests {
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}",
GET_TABLE_LIFECYCLE,
);
assert_action(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
RUN_TABLE_MAINTENANCE,
);
assert_action(
HttpMethod::Post,
"/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
RUN_TABLE_MAINTENANCE,
);
assert_action(
HttpMethod::Post,
"/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import",
rustfs/src/admin/route_registration_test.rs
+40
View File
@@ -397,6 +397,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}",
"/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1",
),
table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
"/analytics/namespaces/sales/tables/orders/maintenance/worker/run",
),
table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
"/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat",
),
table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -535,6 +545,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}",
"/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1",
),
compat_table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
"/analytics/namespaces/sales/tables/orders/maintenance/worker/run",
),
compat_table_route_sample(
Method::POST,
"/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}/heartbeat",
"/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat",
),
compat_table_route_sample(
Method::GET,
"/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export",
@@ -744,6 +764,16 @@ fn test_register_routes_cover_representative_admin_paths() {
Method::GET,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1"),
);
assert_route(
&router,
Method::POST,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/worker/run"),
);
assert_route(
&router,
Method::POST,
&table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat"),
);
assert_route(
&router,
Method::GET,
@@ -856,6 +886,16 @@ fn test_register_routes_cover_representative_admin_paths() {
Method::GET,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1"),
);
assert_route(
&router,
Method::POST,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/worker/run"),
);
assert_route(
&router,
Method::POST,
&compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1/heartbeat"),
);
assert_route(
&router,
Method::GET,
rustfs/src/table_catalog.rs
+600 -9
View File
@@ -84,6 +84,8 @@ const MAINTENANCE_JOB_ALIAS_CURRENT: &str = "current";
const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000;
const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60;
const TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS: u64 = 24 * 60 * 60;
const TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_DEFAULT_SECONDS: u64 = 15 * 60;
const TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_MAX_SECONDS: u64 = 24 * 60 * 60;
const TABLE_COMMIT_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2);
const ICEBERG_MAIN_REF: &str = "main";
const ICEBERG_MIN_SNAPSHOTS_TO_KEEP_PROPERTY: &str = "history.expire.min-snapshots-to-keep";
@@ -274,6 +276,13 @@ pub(crate) struct TableMaintenanceConfig {
pub delete_enabled: bool,
#[serde(rename = "background-enabled")]
pub background_enabled: bool,
#[serde(default, rename = "worker-paused")]
pub worker_paused: bool,
#[serde(
default = "default_table_maintenance_worker_lease_timeout_seconds",
rename = "worker-lease-timeout-seconds"
)]
pub worker_lease_timeout_seconds: u64,
#[serde(default, rename = "max-retry-attempts")]
pub max_retry_attempts: u16,
#[serde(default, rename = "retry-initial-backoff-seconds")]
@@ -293,6 +302,8 @@ impl Default for TableMaintenanceConfig {
retain_recent_metadata_files: 0,
delete_enabled: false,
background_enabled: false,
worker_paused: false,
worker_lease_timeout_seconds: TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_DEFAULT_SECONDS,
max_retry_attempts: 0,
retry_initial_backoff_seconds: 5,
retry_max_backoff_seconds: 300,
@@ -553,6 +564,7 @@ pub(crate) enum TableMetadataMaintenanceJobStatus {
Successful,
Failed,
Disabled,
Paused,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -607,6 +619,26 @@ pub(crate) struct TableMetadataMaintenanceReferencedObjectReport {
pub reasons: Vec<TableMetadataMaintenanceReason>,
}
struct TableMaintenanceHeartbeatRef<'a> {
table_bucket: &'a str,
namespace: &'a str,
table: &'a str,
job_id: &'a str,
lease_id: &'a str,
worker_id: &'a str,
}
struct TableMaintenanceWorkerControlReport<'a> {
table_bucket: &'a str,
namespace: &'a str,
table: &'a str,
worker_id: String,
effective: &'a TableMaintenanceEffectiveConfig,
status: TableMetadataMaintenanceJobStatus,
reason: &'a str,
now: OffsetDateTime,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct TableCatalogExport {
pub table_bucket: TableBucketEntry,
@@ -1597,6 +1629,219 @@ where
.map(|entry| entry.map(|(report, _)| report))
}
pub(crate) async fn run_table_metadata_maintenance_worker_once(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
worker_id: String,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
self.run_table_metadata_maintenance_worker_once_at(table_bucket, namespace, table, worker_id, OffsetDateTime::now_utc())
.await
}
async fn run_table_metadata_maintenance_worker_once_at(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
worker_id: String,
now: OffsetDateTime,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
let effective = self
.get_effective_table_maintenance_config(table_bucket, namespace, table)
.await?;
if !effective.config.background_enabled {
return self
.put_table_metadata_maintenance_worker_control_report(TableMaintenanceWorkerControlReport {
table_bucket,
namespace,
table,
worker_id,
effective: &effective,
status: TableMetadataMaintenanceJobStatus::Disabled,
reason: "background maintenance is disabled",
now,
})
.await;
}
if effective.config.worker_paused {
return self
.put_table_metadata_maintenance_worker_control_report(TableMaintenanceWorkerControlReport {
table_bucket,
namespace,
table,
worker_id,
effective: &effective,
status: TableMetadataMaintenanceJobStatus::Paused,
reason: "background maintenance worker is paused",
now,
})
.await;
}
if let Some(current) = self
.get_table_metadata_maintenance_report(table_bucket, namespace, table, MAINTENANCE_JOB_ALIAS_CURRENT)
.await?
{
if matches!(current.job.status, TableMetadataMaintenanceJobStatus::Running) {
if table_maintenance_job_lease_is_active(&current.job, effective.config.worker_lease_timeout_seconds, now) {
return Ok(current);
}
let mut expired = current;
expired.job.status = TableMetadataMaintenanceJobStatus::Failed;
expired.job.failure_reason = Some("maintenance worker lease expired".to_string());
expired.job.finished_at = Some(maintenance_timestamp(now));
self.put_table_metadata_maintenance_report(&expired).await?;
} else if table_maintenance_job_retry_is_pending(&current.job, now) {
return Ok(current);
}
}
self.run_table_metadata_maintenance_with_config(
table_bucket,
namespace,
table,
effective.config.delete_enabled,
Some(worker_id),
effective,
)
.await
}
pub(crate) async fn heartbeat_table_metadata_maintenance_job(
&self,
table_bucket: &str,
namespace: &str,
table: &str,
job_id: &str,
lease_id: &str,
worker_id: &str,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
self.heartbeat_table_metadata_maintenance_job_at(
TableMaintenanceHeartbeatRef {
table_bucket,
namespace,
table,
job_id,
lease_id,
worker_id,
},
OffsetDateTime::now_utc(),
)
.await
}
async fn heartbeat_table_metadata_maintenance_job_at(
&self,
heartbeat: TableMaintenanceHeartbeatRef<'_>,
now: OffsetDateTime,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
let namespace = parse_namespace_for_store(heartbeat.namespace)?;
let table = parse_table_for_store(heartbeat.table)?;
let table_path = self.paths.table_entry_path(heartbeat.table_bucket, &namespace, &table);
let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?;
let Some(mut report) = self
.get_table_metadata_maintenance_report(
heartbeat.table_bucket,
&namespace.public_name(),
table.as_str(),
MAINTENANCE_JOB_ALIAS_CURRENT,
)
.await?
else {
return Err(TableCatalogStoreError::NotFound(format!(
"maintenance job {}/{}/{}/{}",
heartbeat.table_bucket,
namespace.public_name(),
table.as_str(),
heartbeat.job_id
)));
};
if report.job.job_id != heartbeat.job_id {
return Err(TableCatalogStoreError::Conflict("maintenance job is not current".to_string()));
}
if !matches!(report.job.status, TableMetadataMaintenanceJobStatus::Running) {
return Err(TableCatalogStoreError::Conflict("maintenance job is not running".to_string()));
}
if report.job.lease_id != heartbeat.lease_id {
return Err(TableCatalogStoreError::Conflict("maintenance lease does not match".to_string()));
}
if report.job.worker_id.as_deref() != Some(heartbeat.worker_id) {
return Err(TableCatalogStoreError::Conflict("maintenance worker does not match".to_string()));
}
report.job.heartbeat_at = Some(maintenance_timestamp(now));
self.put_table_metadata_maintenance_report(&report).await?;
Ok(report)
}
async fn put_table_metadata_maintenance_worker_control_report(
&self,
control: TableMaintenanceWorkerControlReport<'_>,
) -> TableCatalogStoreResult<TableMetadataMaintenanceReport> {
let namespace = parse_namespace_for_store(control.namespace)?;
let table = parse_table_for_store(control.table)?;
let table_path = self.paths.table_entry_path(control.table_bucket, &namespace, &table);
let Some((entry, _)) = self.read_entry::<TableEntry>(self.catalog_bucket(), &table_path).await? else {
return Err(TableCatalogStoreError::NotFound(format!(
"table {}/{}/{}",
control.table_bucket,
namespace.public_name(),
table.as_str()
)));
};
let timestamp = maintenance_timestamp(control.now);
let cleanup_watermark_unix_seconds =
(control.now - Duration::seconds(TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS)).unix_timestamp();
let current_metadata_location = entry.metadata_location.clone();
let report = TableMetadataMaintenanceReport {
job: TableMetadataMaintenanceJob {
job_id: Uuid::new_v4().to_string(),
table_bucket: control.table_bucket.to_string(),
namespace: namespace.public_name(),
table: table.as_str().to_string(),
table_id: entry.table_id,
operation: TableMetadataMaintenanceOperation::DryRun,
status: control.status,
failure_reason: Some(control.reason.to_string()),
config_source: control.effective.source,
worker_id: Some(control.worker_id),
lease_id: String::new(),
attempt: 0,
max_retry_attempts: control.effective.config.max_retry_attempts,
next_retry_after: None,
quarantine_enabled: control.effective.config.quarantine_enabled,
quarantine_retention_seconds: control.effective.config.quarantine_retention_seconds,
heartbeat_at: None,
started_at: Some(timestamp.clone()),
finished_at: Some(timestamp),
current_metadata_location: current_metadata_location.clone(),
current_generation: entry.generation,
retain_recent_metadata_files: control.effective.config.retain_recent_metadata_files,
safety_window_seconds: TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS,
cleanup_watermark_unix_seconds,
planned_metadata_file_count: 0,
retained_metadata_file_count: 0,
cleanup_candidate_count: 0,
deletable_metadata_file_count: 0,
deleted_metadata_file_count: 0,
quarantined_object_count: 0,
},
current_metadata_location,
retained_metadata_locations: Vec::new(),
cleanup_candidate_locations: Vec::new(),
deletable_metadata_locations: Vec::new(),
object_reports: Vec::new(),
referenced_object_reports: Vec::new(),
reachability_graph: TableMaintenanceReachabilityGraphReport::default(),
snapshot_expiration: None,
compaction: None,
};
self.put_table_metadata_maintenance_report(&report).await?;
Ok(report)
}
pub(crate) async fn plan_table_snapshot_expiration(
&self,
table_bucket: &str,
@@ -3448,6 +3693,36 @@ fn maintenance_timestamp(now: OffsetDateTime) -> String {
.unwrap_or_else(|_| now.unix_timestamp().to_string())
}
fn default_table_maintenance_worker_lease_timeout_seconds() -> u64 {
TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_DEFAULT_SECONDS
}
fn parse_maintenance_timestamp(timestamp: &str) -> Option<OffsetDateTime> {
OffsetDateTime::parse(timestamp, &time::format_description::well_known::Rfc3339).ok()
}
fn table_maintenance_job_lease_is_active(
job: &TableMetadataMaintenanceJob,
worker_lease_timeout_seconds: u64,
now: OffsetDateTime,
) -> bool {
let Some(heartbeat_at) = job.heartbeat_at.as_deref().and_then(parse_maintenance_timestamp) else {
return false;
};
let timeout_seconds = i64::try_from(worker_lease_timeout_seconds).unwrap_or(i64::MAX);
heartbeat_at.saturating_add(Duration::seconds(timeout_seconds)) > now
}
fn table_maintenance_job_retry_is_pending(job: &TableMetadataMaintenanceJob, now: OffsetDateTime) -> bool {
if !matches!(job.status, TableMetadataMaintenanceJobStatus::Failed) {
return false;
}
let Some(next_retry_after) = job.next_retry_after.as_deref().and_then(parse_maintenance_timestamp) else {
return false;
};
next_retry_after > now
}
fn validate_catalog_entry_version(kind: &str, version: u16) -> TableCatalogStoreResult<()> {
if version != TABLE_CATALOG_ENTRY_VERSION {
return Err(TableCatalogStoreError::Invalid(format!("unsupported {kind} entry version")));
@@ -3466,11 +3741,16 @@ fn validate_table_maintenance_config_version(version: u16) -> TableCatalogStoreR
fn validate_table_maintenance_config(config: &TableMaintenanceConfig) -> TableCatalogStoreResult<()> {
validate_table_maintenance_config_version(config.version)?;
if config.background_enabled {
if config.worker_lease_timeout_seconds == 0 {
return Err(TableCatalogStoreError::Invalid(
"background table maintenance is not supported".to_string(),
"worker-lease-timeout-seconds must be greater than zero".to_string(),
));
}
if config.worker_lease_timeout_seconds > TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_MAX_SECONDS {
return Err(TableCatalogStoreError::Invalid(format!(
"worker-lease-timeout-seconds cannot exceed {TABLE_MAINTENANCE_WORKER_LEASE_TIMEOUT_MAX_SECONDS}"
)));
}
if config.max_retry_attempts > 10 {
return Err(TableCatalogStoreError::Invalid("max-retry-attempts cannot exceed 10".to_string()));
}
@@ -5246,7 +5526,7 @@ mod tests {
}
#[tokio::test]
async fn maintenance_config_rejects_background_enabled_until_worker_exists() {
async fn maintenance_config_accepts_background_enabled_worker_runtime_controls() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend);
let bucket = "analytics";
@@ -5256,7 +5536,7 @@ mod tests {
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
let bucket_err = store
let bucket_config = store
.put_table_bucket_maintenance_config(
bucket,
TableMaintenanceConfig {
@@ -5264,14 +5544,18 @@ mod tests {
retain_recent_metadata_files: 1,
delete_enabled: false,
background_enabled: true,
worker_paused: true,
worker_lease_timeout_seconds: 60,
..Default::default()
},
)
.await
.unwrap_err();
assert_matches!(bucket_err, TableCatalogStoreError::Invalid(_));
.expect("background maintenance bucket config should persist");
assert!(bucket_config.background_enabled);
assert!(bucket_config.worker_paused);
assert_eq!(bucket_config.worker_lease_timeout_seconds, 60);
let table_err = store
let table_config = store
.put_table_maintenance_config(
bucket,
"sales",
@@ -5281,12 +5565,16 @@ mod tests {
retain_recent_metadata_files: 1,
delete_enabled: false,
background_enabled: true,
worker_paused: false,
worker_lease_timeout_seconds: 120,
..Default::default()
},
)
.await
.unwrap_err();
assert_matches!(table_err, TableCatalogStoreError::Invalid(_));
.expect("background maintenance table config should persist");
assert!(table_config.background_enabled);
assert!(!table_config.worker_paused);
assert_eq!(table_config.worker_lease_timeout_seconds, 120);
}
#[tokio::test]
@@ -5315,6 +5603,7 @@ mod tests {
retry_max_backoff_seconds: 60,
quarantine_enabled: true,
quarantine_retention_seconds: 86_400,
..Default::default()
},
)
.await
@@ -5505,6 +5794,7 @@ mod tests {
retry_max_backoff_seconds: 30,
quarantine_enabled: true,
quarantine_retention_seconds: 86_400,
..Default::default()
},
)
.await
@@ -5544,6 +5834,307 @@ mod tests {
assert_eq!(latest.job.status, TableMetadataMaintenanceJobStatus::Failed);
}
#[tokio::test]
async fn maintenance_worker_run_skips_when_background_is_disabled() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend.seed_object(bucket, &old, b"{}".to_vec()).await;
backend
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
let report = store
.run_table_metadata_maintenance_worker_once(bucket, "sales", "orders", "worker-a".to_string())
.await
.expect("disabled background worker tick should report a safe no-op");
assert_eq!(report.job.status, TableMetadataMaintenanceJobStatus::Disabled);
assert_eq!(report.job.worker_id.as_deref(), Some("worker-a"));
assert_eq!(report.job.deleted_metadata_file_count, 0);
assert!(backend.object_exists(bucket, &old).await.unwrap());
}
#[tokio::test]
async fn maintenance_worker_run_honors_paused_config() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend.seed_object(bucket, &old, b"{}".to_vec()).await;
backend
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
store
.put_table_maintenance_config(
bucket,
"sales",
"orders",
TableMaintenanceConfig {
version: TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 0,
delete_enabled: true,
background_enabled: true,
worker_paused: true,
..Default::default()
},
)
.await
.expect("paused background maintenance config should persist");
let report = store
.run_table_metadata_maintenance_worker_once(bucket, "sales", "orders", "worker-a".to_string())
.await
.expect("paused worker tick should report a safe no-op");
assert_eq!(report.job.status, TableMetadataMaintenanceJobStatus::Paused);
assert_eq!(report.job.operation, TableMetadataMaintenanceOperation::DryRun);
assert_eq!(report.job.deleted_metadata_file_count, 0);
assert!(backend.object_exists(bucket, &old).await.unwrap());
}
#[tokio::test]
async fn maintenance_worker_run_defers_until_retry_after() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100);
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend.seed_object(bucket, &old, b"{}".to_vec()).await;
backend
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
store
.put_table_maintenance_config(
bucket,
"sales",
"orders",
TableMaintenanceConfig {
version: TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 0,
delete_enabled: false,
background_enabled: true,
max_retry_attempts: 2,
retry_initial_backoff_seconds: 60,
retry_max_backoff_seconds: 60,
..Default::default()
},
)
.await
.expect("retry-enabled maintenance config should persist");
let mut failed = store
.run_table_metadata_maintenance(bucket, "sales", "orders", true, Some("worker-a".to_string()))
.await
.expect("delete failure should be recorded when delete is disabled");
failed.job.next_retry_after = Some(maintenance_timestamp(now + Duration::seconds(30)));
store
.put_table_metadata_maintenance_report(&failed)
.await
.expect("failed retry report should be seeded");
let deferred = store
.run_table_metadata_maintenance_worker_once_at(bucket, "sales", "orders", "worker-b".to_string(), now)
.await
.expect("worker tick should defer while retry backoff is active");
assert_eq!(deferred.job.job_id, failed.job.job_id);
assert_eq!(deferred.job.status, TableMetadataMaintenanceJobStatus::Failed);
assert_eq!(deferred.job.worker_id.as_deref(), Some("worker-a"));
assert!(backend.object_exists(bucket, &old).await.unwrap());
}
#[tokio::test]
async fn maintenance_worker_run_backpressures_active_running_job() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100);
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
store
.put_table_maintenance_config(
bucket,
"sales",
"orders",
TableMaintenanceConfig {
version: TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 0,
delete_enabled: false,
background_enabled: true,
worker_lease_timeout_seconds: 300,
..Default::default()
},
)
.await
.expect("background maintenance config should persist");
let mut running = store
.plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
.await
.expect("maintenance report should be planned");
running.job.status = TableMetadataMaintenanceJobStatus::Running;
running.job.worker_id = Some("worker-a".to_string());
running.job.lease_id = "lease-a".to_string();
running.job.heartbeat_at = Some(maintenance_timestamp(now - Duration::seconds(10)));
store
.put_table_metadata_maintenance_report(&running)
.await
.expect("running maintenance report should be seeded");
let report = store
.run_table_metadata_maintenance_worker_once_at(bucket, "sales", "orders", "worker-b".to_string(), now)
.await
.expect("worker tick should return the active running job");
assert_eq!(report.job.job_id, running.job.job_id);
assert_eq!(report.job.status, TableMetadataMaintenanceJobStatus::Running);
assert_eq!(report.job.worker_id.as_deref(), Some("worker-a"));
}
#[tokio::test]
async fn maintenance_worker_run_recovers_expired_running_job() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let old = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
let now = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1000);
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current.clone()).await;
backend.seed_object(bucket, &old, b"{}".to_vec()).await;
backend
.seed_object(bucket, &current, br#"{"metadata-log":[]}"#.to_vec())
.await;
store
.put_table_maintenance_config(
bucket,
"sales",
"orders",
TableMaintenanceConfig {
version: TABLE_MAINTENANCE_CONFIG_VERSION,
retain_recent_metadata_files: 0,
delete_enabled: false,
background_enabled: true,
worker_lease_timeout_seconds: 60,
..Default::default()
},
)
.await
.expect("background maintenance config should persist");
let mut running = store
.plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
.await
.expect("maintenance report should be planned");
let expired_job_id = running.job.job_id.clone();
running.job.status = TableMetadataMaintenanceJobStatus::Running;
running.job.worker_id = Some("worker-a".to_string());
running.job.lease_id = "lease-a".to_string();
running.job.heartbeat_at = Some(maintenance_timestamp(now - Duration::seconds(120)));
store
.put_table_metadata_maintenance_report(&running)
.await
.expect("expired running maintenance report should be seeded");
let report = store
.run_table_metadata_maintenance_worker_once_at(bucket, "sales", "orders", "worker-b".to_string(), now)
.await
.expect("worker tick should recover expired running job and run again");
assert_ne!(report.job.job_id, expired_job_id);
assert_eq!(report.job.status, TableMetadataMaintenanceJobStatus::Successful);
assert_eq!(report.job.worker_id.as_deref(), Some("worker-b"));
let expired = store
.get_table_metadata_maintenance_report(bucket, "sales", "orders", &expired_job_id)
.await
.expect("expired job lookup should succeed")
.expect("expired job should remain addressable");
assert_eq!(expired.job.status, TableMetadataMaintenanceJobStatus::Failed);
assert!(
expired
.job
.failure_reason
.as_deref()
.is_some_and(|reason| reason.contains("lease expired"))
);
}
#[tokio::test]
async fn maintenance_worker_heartbeat_updates_current_running_job() {
let backend = TestCatalogObjectBackend::default();
let store = ObjectTableCatalogStore::new(backend.clone());
let bucket = "analytics";
let namespace = Namespace::parse("sales").unwrap();
let table = IdentifierSegment::parse("orders").unwrap();
let current = default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
let first = OffsetDateTime::UNIX_EPOCH + Duration::seconds(100);
let second = OffsetDateTime::UNIX_EPOCH + Duration::seconds(130);
seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await;
backend
.seed_object(
bucket,
&default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"),
br#"{"metadata-log":[]}"#.to_vec(),
)
.await;
let mut running = store
.plan_table_metadata_maintenance(bucket, "sales", "orders", 0)
.await
.expect("maintenance report should be planned");
running.job.status = TableMetadataMaintenanceJobStatus::Running;
running.job.worker_id = Some("worker-a".to_string());
running.job.lease_id = "lease-a".to_string();
running.job.heartbeat_at = Some(maintenance_timestamp(first));
let job_id = running.job.job_id.clone();
store
.put_table_metadata_maintenance_report(&running)
.await
.expect("running maintenance report should be seeded");
let heartbeat = store
.heartbeat_table_metadata_maintenance_job_at(
TableMaintenanceHeartbeatRef {
table_bucket: bucket,
namespace: "sales",
table: "orders",
job_id: &job_id,
lease_id: "lease-a",
worker_id: "worker-a",
},
second,
)
.await
.expect("heartbeat should update the current running job");
assert_eq!(heartbeat.job.job_id, job_id);
assert_eq!(heartbeat.job.status, TableMetadataMaintenanceJobStatus::Running);
assert_eq!(heartbeat.job.heartbeat_at.as_deref(), Some(maintenance_timestamp(second).as_str()));
}
#[tokio::test]
async fn maintenance_reachability_reports_manifest_lists_as_manual_review() {
let backend = TestCatalogObjectBackend::default();
scripts/table-catalog/README.md
+2 -2
View File
@@ -149,10 +149,10 @@ Unsupported behavior is documented instead of hidden behind internal errors. The
current unsupported inventory is:
- credential vending: automated after table bootstrap with exact-prefix validation and a data-plane scope probe; full no-long-term-data-credential bootstrap is not claimed
- background maintenance worker: unsupported; maintenance job reports expose retry/quarantine policy fields for future workers
- background maintenance worker: controlled run-once and heartbeat endpoints are registered; continuous in-process scheduling is not claimed
- manifest/data reachability cleanup: fail-closed reachability graph reporting only; metadata cleanup must not delete manifest, data, or delete files
- snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports
- automatic maintenance scheduling: unsupported
- automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint; built-in periodic scheduling is not claimed
- compaction rewrite: unsupported; planning reports fail closed until manifest Avro reading and rewrite support are implemented
- Iceberg views: stable unsupported routes are registered and return explicit unsupported JSON
- external catalog bridges: metadata import/register is supported, but Polaris/Glue/DLF/Hive synchronization is unsupported
scripts/table-catalog/pyiceberg_smoke.py
+3 -2
View File
@@ -133,9 +133,10 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
},
{
"capability": "background-maintenance-worker",
"status": "unsupported",
"status": "controlled-run-once-supported",
"roadmap_area": "maintenance-worker",
"expected_behavior": "manual maintenance APIs expose retry/quarantine policy fields in reports, but background-enabled maintenance is rejected until a worker owns scheduling",
"catalog_endpoint": "POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/maintenance/worker/run",
"expected_behavior": "background-enabled maintenance can be driven by the worker run endpoint with current-job backpressure, retry deferral, lease expiry recovery, and heartbeat updates; built-in periodic scheduling is not claimed",
},
{
"capability": "manifest-data-reachability-cleanup",