From 7ce98380278a80eb2024f6fcb13d2a44df76e531 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 15 Jun 2026 18:42:29 +0800 Subject: [PATCH] feat(table-catalog): add maintenance worker runtime (#3480) --- rustfs/src/admin/handlers/table_catalog.rs | 153 ++++- rustfs/src/admin/route_policy.rs | 36 +- rustfs/src/admin/route_registration_test.rs | 40 ++ rustfs/src/table_catalog.rs | 609 +++++++++++++++++++- scripts/table-catalog/README.md | 4 +- scripts/table-catalog/pyiceberg_smoke.py | 5 +- 6 files changed, 816 insertions(+), 31 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 02e928ac7..6c4329fd8 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -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, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct TableMaintenanceWorkerRunRequest { + #[serde(default, rename = "worker-id")] + worker_id: Option, +} + +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, 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, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?; + let request = read_json_body::(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, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let job = job_id_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::RunTableMaintenanceAction).await?; + let request = read_json_body::(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::(); assert_operation::(); assert_operation::(); + assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -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::(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, diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 840fd8904..6485a1c9e 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -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", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index db9aa3691..14b355d31 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -397,6 +397,16 @@ fn expected_admin_route_matrix() -> Vec { "/{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 { "/{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, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 24e1d2966..4a7567127 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -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, } +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 { + 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 { + 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(¤t.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(¤t.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 { + 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 { + 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 { + 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::(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::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, ¤t, 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, ¤t, 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, ¤t, 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, ¤t, 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, ¤t, 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(); diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index b5c1d7068..a3211609c 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -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 diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 47f72a509..837891a02 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -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",