From a305160087c3639b68f6226e55dea98217e2b072 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 1 Jul 2026 22:31:19 +0800 Subject: [PATCH] feat(table-catalog): add maintenance scheduler guardrails (#4123) --- docs/architecture/s3-tables-support-matrix.md | 3 +- rustfs/src/admin/handlers/table_catalog.rs | 32 ++ rustfs/src/admin/route_policy.rs | 24 +- rustfs/src/admin/route_registration_test.rs | 20 + rustfs/src/table_catalog.rs | 450 +++++++++++++++++- scripts/table-catalog/README.md | 9 +- scripts/table-catalog/pyiceberg_smoke.py | 11 +- scripts/table-catalog/test_pyiceberg_smoke.py | 7 + 8 files changed, 536 insertions(+), 20 deletions(-) diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 8f272c1ec..a505e3e02 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -87,6 +87,7 @@ catalog extension. | Snapshot expiration commit | Preview / controlled | Can manually commit safe snapshot expiration through the catalog. Stale plans fail closed. | | Manifest/data/delete reachability cleanup | Supported | Reads manifest-list and manifest Avro references, reports reachable objects, and deletes only unreferenced table objects that pass the safety window. | | Maintenance worker run endpoint | Preview / controlled | Supports run-once execution, current-job backpressure, retry deferral, lease expiry recovery, and heartbeat updates. | +| Maintenance scheduler guardrails | Preview / controlled | Exposes disabled, paused, ready, active-job backpressure, retry deferral, quarantine boundary, recommended actions, and recent maintenance job audit timeline state for external schedulers and operators. | | Compaction planning | Preview / controlled | Plans partition-local binpack candidates for Parquet files and does not mix data files from different partition directories in one rewrite group. | | Compaction commit | Preview / controlled | Can commit a safe partition-local Parquet rewrite through the catalog. | | Built-in periodic scheduler | Not claimed | Operators can trigger worker runs, but continuous in-process scheduling is not claimed. | @@ -158,7 +159,7 @@ RustFS does not currently claim: - full MinIO AIStor Tables private extension parity - full Cloudflare R2 Data Catalog interoperability - full Alibaba OSS Tables interoperability -- built-in periodic maintenance scheduling +- built-in periodic maintenance scheduling; external schedulers can inspect scheduler guardrails, but RustFS does not claim a continuous in-process scheduler - active-active multi-region table writes - multi-table transactions - no-long-term-data-credential table bootstrap diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 1138fc58f..ad030c7ee 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -135,6 +135,7 @@ 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}", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", "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", @@ -179,6 +180,7 @@ 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 GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER: GetTableMaintenanceSchedulerHandler = GetTableMaintenanceSchedulerHandler {}; static RUN_TABLE_MAINTENANCE_WORKER_HANDLER: RunTableMaintenanceWorkerHandler = RunTableMaintenanceWorkerHandler {}; static HEARTBEAT_TABLE_MAINTENANCE_JOB_HANDLER: HeartbeatTableMaintenanceJobHandler = HeartbeatTableMaintenanceJobHandler {}; static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {}; @@ -942,6 +944,11 @@ 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::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/scheduler").as_str(), + AdminOperation(&GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER), + )?; r.insert( Method::POST, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/worker/run").as_str(), @@ -5201,6 +5208,26 @@ impl Operation for GetTableMaintenanceJobHandler { } } +pub struct GetTableMaintenanceSchedulerHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableMaintenanceSchedulerHandler { + 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::GetTableLifecycleAction).await?; + ensure_table_bucket_enabled(&warehouse).await?; + let store = table_catalog_store()?; + let response = store + .get_table_maintenance_scheduler_report(&warehouse, &namespace.public_name(), &table) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + pub struct RunTableMaintenanceWorkerHandler {} #[async_trait::async_trait] @@ -5641,6 +5668,7 @@ mod tests { ("GetTableMaintenanceConfigHandler", "AdminAction::GetTableLifecycleAction"), ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), + ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"), @@ -5694,6 +5722,7 @@ mod tests { ("GetTableMaintenanceConfigHandler", "AdminAction::GetTableLifecycleAction"), ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), + ("GetTableMaintenanceSchedulerHandler", "AdminAction::GetTableLifecycleAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"), @@ -5749,6 +5778,7 @@ mod tests { "GetTableMaintenanceConfigHandler", "PutTableMaintenanceConfigHandler", "GetTableMaintenanceJobHandler", + "GetTableMaintenanceSchedulerHandler", "RunTableMaintenanceWorkerHandler", "HeartbeatTableMaintenanceJobHandler", "ExportTableCatalogHandler", @@ -5857,6 +5887,7 @@ mod tests { let _: &GetTableMaintenanceConfigHandler = &GET_TABLE_MAINTENANCE_CONFIG_HANDLER; let _: &PutTableMaintenanceConfigHandler = &PUT_TABLE_MAINTENANCE_CONFIG_HANDLER; let _: &GetTableMaintenanceJobHandler = &GET_TABLE_MAINTENANCE_JOB_HANDLER; + let _: &GetTableMaintenanceSchedulerHandler = &GET_TABLE_MAINTENANCE_SCHEDULER_HANDLER; let _: &ExportTableCatalogHandler = &EXPORT_TABLE_CATALOG_HANDLER; let _: &ImportTableCatalogHandler = &IMPORT_TABLE_CATALOG_HANDLER; let _: &ExternalCatalogBridgeHandler = &EXTERNAL_CATALOG_BRIDGE_HANDLER; @@ -5894,6 +5925,7 @@ mod tests { assert_operation::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 04ad809db..4d5538e97 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -829,6 +829,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_LIFECYCLE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -1076,6 +1082,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_LIFECYCLE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -1318,7 +1330,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(), 84); + assert_eq!(table_specs.count(), 86); 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); @@ -1443,6 +1455,16 @@ mod tests { "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", GET_TABLE_LIFECYCLE, ); + assert_action( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + GET_TABLE_LIFECYCLE, + ); + assert_action( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + GET_TABLE_LIFECYCLE, + ); assert_action( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 1a350735d..397550e49 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -419,6 +419,11 @@ 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::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + "/analytics/namespaces/sales/tables/orders/maintenance/scheduler", + ), table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -593,6 +598,11 @@ 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::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + "/analytics/namespaces/sales/tables/orders/maintenance/scheduler", + ), compat_table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/worker/run", @@ -839,6 +849,11 @@ 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::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler"), + ); assert_route( &router, Method::POST, @@ -986,6 +1001,11 @@ 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::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/scheduler"), + ); assert_route( &router, Method::POST, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 333c9a311..d72f00613 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -111,6 +111,7 @@ 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_MAINTENANCE_SCHEDULER_AUDIT_LIMIT: usize = 10; const TABLE_MAINTENANCE_DELETE_DISABLED_REASON: &str = "metadata delete is disabled by maintenance config"; const TABLE_COMMIT_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2); const ICEBERG_MAIN_REF: &str = "main"; @@ -510,6 +511,61 @@ pub(crate) struct TableMaintenanceEffectiveConfig { pub source: TableMaintenanceConfigSource, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableMaintenanceSchedulerStatus { + Ready, + Disabled, + Paused, + Backpressured, + RetryDeferred, + Quarantined, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMaintenanceSchedulerReport { + pub table_bucket: String, + pub namespace: String, + pub table: String, + pub table_id: String, + pub status: TableMaintenanceSchedulerStatus, + pub config_source: TableMaintenanceConfigSource, + pub background_enabled: bool, + pub worker_paused: bool, + pub delete_enabled: bool, + pub worker_lease_timeout_seconds: u64, + pub max_retry_attempts: u16, + pub retry_initial_backoff_seconds: u64, + pub retry_max_backoff_seconds: u64, + pub recommended_actions: Vec, + pub current_job: Option, + pub quarantine: TableMaintenanceSchedulerQuarantineBoundary, + pub audit_timeline: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMaintenanceSchedulerQuarantineBoundary { + pub enabled: bool, + pub active: bool, + pub retention_seconds: u64, + pub quarantined_object_count: usize, + pub source_job_id: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMaintenanceSchedulerJobSummary { + pub job_id: String, + pub operation: TableMetadataMaintenanceOperation, + pub status: TableMetadataMaintenanceJobStatus, + pub worker_id: Option, + pub attempt: u16, + pub started_at: Option, + pub finished_at: Option, + pub heartbeat_at: Option, + pub next_retry_after: Option, + pub recommended_actions: Vec, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct TableMetadataMaintenanceJob { pub job_id: String, @@ -795,6 +851,7 @@ pub(crate) enum TableMetadataMaintenanceJobStatus { pub(crate) enum TableMaintenanceRecommendedAction { NoActionRequired, ReviewAndRunDelete, + ReviewQuarantine, EnableDelete, EnableBackgroundMaintenance, ResumeMaintenanceWorker, @@ -1691,6 +1748,21 @@ impl TableCatalogObjectPaths { ) } + pub fn table_maintenance_jobs_prefix( + &self, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + table_id: &str, + ) -> String { + format!( + "{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_JOB_ROOT}/", + self.table_entries_prefix(table_bucket, namespace), + table.as_str(), + table_catalog_path_hash(table_id) + ) + } + pub fn table_maintenance_latest_job_path( &self, table_bucket: &str, @@ -3823,6 +3895,132 @@ where .map(|entry| entry.map(|(report, _)| table_maintenance_report_with_recommended_actions(report))) } + pub(crate) async fn get_table_maintenance_scheduler_report( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + self.get_table_maintenance_scheduler_report_at(table_bucket, namespace, table, OffsetDateTime::now_utc()) + .await + } + + async fn get_table_maintenance_scheduler_report_at( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + now: OffsetDateTime, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + let effective = self + .get_effective_table_maintenance_config(table_bucket, &namespace.public_name(), table.as_str()) + .await?; + let current = self + .get_table_metadata_maintenance_report( + table_bucket, + &namespace.public_name(), + table.as_str(), + MAINTENANCE_JOB_ALIAS_CURRENT, + ) + .await?; + let reports = self + .list_table_metadata_maintenance_audit_reports(table_bucket, &namespace, &table, &entry.table_id) + .await?; + let quarantine = table_maintenance_scheduler_quarantine_boundary(&effective.config, &reports); + let mut recommended_actions = Vec::new(); + + let status = if !effective.config.background_enabled { + push_unique_maintenance_action( + &mut recommended_actions, + TableMaintenanceRecommendedAction::EnableBackgroundMaintenance, + ); + TableMaintenanceSchedulerStatus::Disabled + } else if effective.config.worker_paused { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::ResumeMaintenanceWorker); + TableMaintenanceSchedulerStatus::Paused + } else if let Some(current) = current.as_ref() + && matches!(current.job.status, TableMetadataMaintenanceJobStatus::Running) + && table_maintenance_job_lease_is_active(¤t.job, effective.config.worker_lease_timeout_seconds, now) + { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::WaitForActiveWorker); + TableMaintenanceSchedulerStatus::Backpressured + } else if let Some(current) = current.as_ref() + && table_maintenance_job_retry_is_pending(¤t.job, now) + { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::WaitForRetryBackoff); + TableMaintenanceSchedulerStatus::RetryDeferred + } else if quarantine.active { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::ReviewQuarantine); + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::InvestigateFailure); + TableMaintenanceSchedulerStatus::Quarantined + } else { + push_unique_maintenance_action(&mut recommended_actions, TableMaintenanceRecommendedAction::NoActionRequired); + TableMaintenanceSchedulerStatus::Ready + }; + + Ok(TableMaintenanceSchedulerReport { + table_bucket: table_bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + table_id: entry.table_id, + status, + config_source: effective.source, + background_enabled: effective.config.background_enabled, + worker_paused: effective.config.worker_paused, + delete_enabled: effective.config.delete_enabled, + worker_lease_timeout_seconds: effective.config.worker_lease_timeout_seconds, + max_retry_attempts: effective.config.max_retry_attempts, + retry_initial_backoff_seconds: effective.config.retry_initial_backoff_seconds, + retry_max_backoff_seconds: effective.config.retry_max_backoff_seconds, + recommended_actions, + current_job: current.as_ref().map(table_maintenance_scheduler_job_summary), + quarantine, + audit_timeline: reports.iter().map(table_maintenance_scheduler_job_summary).collect(), + }) + } + + async fn list_table_metadata_maintenance_audit_reports( + &self, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + table_id: &str, + ) -> TableCatalogStoreResult> { + let jobs_prefix = self + .paths + .table_maintenance_jobs_prefix(table_bucket, namespace, table, table_id); + let mut reports = Vec::new(); + for object in self.backend.list_objects(self.catalog_bucket(), &jobs_prefix).await? { + if !object.ends_with(".json") { + continue; + } + if let Some((report, _)) = self + .read_entry::(self.catalog_bucket(), &object) + .await? + { + reports.push(table_maintenance_report_with_recommended_actions(report)); + } + } + reports.sort_by(|left, right| { + table_maintenance_report_order_timestamp(right) + .cmp(&table_maintenance_report_order_timestamp(left)) + .then_with(|| left.job.job_id.cmp(&right.job.job_id)) + }); + reports.truncate(TABLE_MAINTENANCE_SCHEDULER_AUDIT_LIMIT); + Ok(reports) + } + pub(crate) async fn run_table_metadata_maintenance_worker_once( &self, table_bucket: &str, @@ -5933,6 +6131,22 @@ where } } + pub(crate) async fn get_table_maintenance_scheduler_report( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(store) => { + store + .get_table_maintenance_scheduler_report(table_bucket, namespace, table) + .await + } + Self::DurableStrong(_) => Err(Self::unsupported_for_durable_strong("table maintenance scheduler")), + } + } + pub(crate) async fn run_table_metadata_maintenance_worker_once( &self, table_bucket: &str, @@ -8479,6 +8693,9 @@ fn table_maintenance_recommended_actions(job: &TableMetadataMaintenanceJob) -> V { actions.push(TableMaintenanceRecommendedAction::EnableDelete); } + if job.quarantine_enabled && job.quarantined_object_count > 0 { + actions.push(TableMaintenanceRecommendedAction::ReviewQuarantine); + } if job.next_retry_after.is_some() { actions.push(TableMaintenanceRecommendedAction::WaitForRetryBackoff); } @@ -8496,6 +8713,56 @@ fn table_maintenance_recommended_actions(job: &TableMetadataMaintenanceJob) -> V actions } +fn push_unique_maintenance_action( + actions: &mut Vec, + action: TableMaintenanceRecommendedAction, +) { + if !actions.contains(&action) { + actions.push(action); + } +} + +fn table_maintenance_report_order_timestamp(report: &TableMetadataMaintenanceReport) -> String { + report + .job + .finished_at + .clone() + .or_else(|| report.job.heartbeat_at.clone()) + .or_else(|| report.job.started_at.clone()) + .unwrap_or_default() +} + +fn table_maintenance_scheduler_job_summary(report: &TableMetadataMaintenanceReport) -> TableMaintenanceSchedulerJobSummary { + TableMaintenanceSchedulerJobSummary { + job_id: report.job.job_id.clone(), + operation: report.job.operation.clone(), + status: report.job.status.clone(), + worker_id: report.job.worker_id.clone(), + attempt: report.job.attempt, + started_at: report.job.started_at.clone(), + finished_at: report.job.finished_at.clone(), + heartbeat_at: report.job.heartbeat_at.clone(), + next_retry_after: report.job.next_retry_after.clone(), + recommended_actions: report.job.recommended_actions.clone(), + } +} + +fn table_maintenance_scheduler_quarantine_boundary( + config: &TableMaintenanceConfig, + reports: &[TableMetadataMaintenanceReport], +) -> TableMaintenanceSchedulerQuarantineBoundary { + let source = reports + .iter() + .find(|report| report.job.quarantine_enabled && report.job.quarantined_object_count > 0); + TableMaintenanceSchedulerQuarantineBoundary { + enabled: config.quarantine_enabled, + active: source.is_some(), + retention_seconds: source.map_or(config.quarantine_retention_seconds, |report| report.job.quarantine_retention_seconds), + quarantined_object_count: source.map_or(0, |report| report.job.quarantined_object_count), + source_job_id: source.map(|report| report.job.job_id.clone()), + } +} + fn refresh_table_maintenance_report_recommended_actions(report: &mut TableMetadataMaintenanceReport) { report.job.recommended_actions = table_maintenance_recommended_actions(&report.job); } @@ -10492,8 +10759,8 @@ mod tests { 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 namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; @@ -10518,8 +10785,8 @@ mod tests { 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 namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; @@ -10565,8 +10832,8 @@ mod tests { 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 namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; @@ -10667,8 +10934,8 @@ mod tests { let backend = TestCatalogObjectBackend::default(); let store = ObjectTableCatalogStore::new(backend); let bucket = "analytics"; - let namespace = Namespace::parse("sales").unwrap(); - let table = IdentifierSegment::parse("orders").unwrap(); + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); let prefix = "tables/shared-table/"; let index_path = store.paths.warehouse_index_entry_path(bucket, prefix); @@ -10713,8 +10980,8 @@ mod tests { 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 namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); let bucket_entry = test_bucket_entry(bucket); let namespace_entry = test_namespace_entry(bucket, &namespace); @@ -10773,8 +11040,8 @@ mod tests { 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 namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); let bucket_entry = test_bucket_entry(bucket); let namespace_entry = test_namespace_entry(bucket, &namespace); @@ -11831,6 +12098,165 @@ mod tests { assert!(backend.object_exists(bucket, &old).await.unwrap()); } + #[tokio::test] + async fn maintenance_scheduler_report_marks_disabled_default() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + 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, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + + let report = store + .get_table_maintenance_scheduler_report_at( + bucket, + "sales", + "orders", + OffsetDateTime::UNIX_EPOCH + Duration::seconds(100), + ) + .await + .expect("scheduler report should load"); + + assert_eq!(report.status, TableMaintenanceSchedulerStatus::Disabled); + assert_eq!(report.config_source, TableMaintenanceConfigSource::Default); + assert!(!report.background_enabled); + assert_eq!( + report.recommended_actions, + vec![TableMaintenanceRecommendedAction::EnableBackgroundMaintenance] + ); + assert!(report.current_job.is_none()); + assert!(report.audit_timeline.is_empty()); + assert!(!report.quarantine.active); + } + + #[tokio::test] + async fn maintenance_scheduler_report_surfaces_active_backpressure_and_audit_timeline() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + 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, + 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 + .get_table_maintenance_scheduler_report_at(bucket, "sales", "orders", now) + .await + .expect("scheduler report should load"); + + assert_eq!(report.status, TableMaintenanceSchedulerStatus::Backpressured); + assert_eq!( + report.current_job.as_ref().map(|job| job.job_id.as_str()), + Some(running.job.job_id.as_str()) + ); + assert_eq!(report.recommended_actions, vec![TableMaintenanceRecommendedAction::WaitForActiveWorker]); + assert_eq!(report.audit_timeline.len(), 1); + assert_eq!(report.audit_timeline[0].job_id, running.job.job_id); + assert_eq!(report.audit_timeline[0].status, TableMetadataMaintenanceJobStatus::Running); + assert_eq!(report.audit_timeline[0].worker_id.as_deref(), Some("worker-a")); + } + + #[tokio::test] + async fn maintenance_scheduler_report_surfaces_quarantine_boundary() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + 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, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + background_enabled: true, + quarantine_enabled: true, + quarantine_retention_seconds: 86_400, + ..Default::default() + }, + ) + .await + .expect("background maintenance config should persist"); + let mut failed = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .expect("maintenance report should be planned"); + failed.job.status = TableMetadataMaintenanceJobStatus::Failed; + failed.job.failure_reason = Some("quarantine retained failed cleanup candidates".to_string()); + failed.job.quarantine_enabled = true; + failed.job.quarantine_retention_seconds = 86_400; + failed.job.quarantined_object_count = 2; + failed.job.finished_at = Some(maintenance_timestamp(OffsetDateTime::UNIX_EPOCH + Duration::seconds(90))); + store + .put_table_metadata_maintenance_report(&failed) + .await + .expect("failed maintenance report should be seeded"); + + let report = store + .get_table_maintenance_scheduler_report_at( + bucket, + "sales", + "orders", + OffsetDateTime::UNIX_EPOCH + Duration::seconds(100), + ) + .await + .expect("scheduler report should load"); + + assert_eq!(report.status, TableMaintenanceSchedulerStatus::Quarantined); + assert!(report.quarantine.active); + assert_eq!(report.quarantine.retention_seconds, 86_400); + assert_eq!(report.quarantine.quarantined_object_count, 2); + assert_eq!(report.quarantine.source_job_id.as_deref(), Some(failed.job.job_id.as_str())); + assert!( + report + .recommended_actions + .contains(&TableMaintenanceRecommendedAction::ReviewQuarantine) + ); + } + #[tokio::test] async fn maintenance_worker_run_defers_until_retry_after() { let backend = TestCatalogObjectBackend::default(); diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 86c076a88..51a6df0e2 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -229,8 +229,9 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces: `main` cannot be deleted - Iceberg views support basic create, list, load, replace, existence check, and drop routes with persisted view metadata and view-scoped authorization -- metadata maintenance supports safe dry-run planning and controlled worker - execution checks +- metadata maintenance supports safe dry-run planning, controlled worker + execution checks, and a scheduler status report with disabled, paused, + backpressure, retry, quarantine, and audit-timeline state - catalog diagnostics exposes the table recovery and consistency state used by operators - catalog export and diagnostics expose the current catalog backing manifest, @@ -333,10 +334,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: controlled run-once and heartbeat endpoints are registered; continuous in-process scheduling is not claimed +- background maintenance worker: controlled run-once, heartbeat, and scheduler status endpoints are registered; disabled/paused/backpressure/retry/quarantine/audit-timeline state is machine-readable; continuous in-process scheduling is not claimed - manifest/data reachability cleanup: metadata maintenance reads manifest-list and manifest Avro references, reports manifest/data/delete reachability, and deletes only unreferenced table objects that pass the safety window - snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports -- automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint; built-in periodic scheduling is not claimed +- automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint and scheduler status report; built-in periodic scheduling is not claimed - compaction rewrite: controlled run-once support for partition-local Parquet binpack through metadata maintenance; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed - row-level delete/update/merge commits: standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item - external catalog bridges: metadata import/register and operator-supplied metadata pointer sync are supported for Polaris/Glue/DLF/Hive identity boundaries; online vendor SDK polling and policy mirroring are not claimed diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 10783d4a5..4c655916b 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -182,8 +182,8 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [ "capability": "background-maintenance-worker", "status": "controlled-run-once-supported", "roadmap_area": "maintenance-worker", - "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", + "catalog_endpoint": "GET /v1/{prefix}/namespaces/{namespace}/tables/{table}/maintenance/scheduler", + "expected_behavior": "background-enabled maintenance can be driven by the worker run endpoint and inspected through the scheduler status endpoint with disabled/paused state, current-job backpressure, retry deferral, quarantine boundary, audit timeline, lease expiry recovery, and heartbeat updates; built-in periodic scheduling is not claimed", }, { "capability": "manifest-data-reachability-cleanup", @@ -1049,6 +1049,7 @@ def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: config_path = table_endpoint_path(args, "/maintenance/config") + scheduler_path = table_endpoint_path(args, "/maintenance/scheduler") signed_rest_request(args, deps, "PUT", config_path, default_maintenance_config()) config = signed_rest_request(args, deps, "GET", config_path) if config.get("version") != TABLE_MAINTENANCE_CONFIG_VERSION: @@ -1065,6 +1066,12 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None: if not isinstance(job, dict) or not job.get("job-id"): raise RuntimeError("maintenance metadata endpoint did not return a job id") signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}")) + scheduler = signed_rest_request(args, deps, "GET", scheduler_path) + expected_scheduler_statuses = {"READY", "DISABLED", "PAUSED", "BACKPRESSURED", "RETRY_DEFERRED", "QUARANTINED"} + if scheduler.get("status") not in expected_scheduler_statuses: + raise RuntimeError("maintenance scheduler endpoint did not return a stable scheduler status") + if "audit_timeline" not in scheduler: + raise RuntimeError("maintenance scheduler endpoint did not return an audit timeline") worker = signed_rest_request(args, deps, "POST", table_endpoint_path(args, "/maintenance/worker/run"), {}) worker_job = worker.get("job") diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index 5ea7a067e..496ed8fa4 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -294,6 +294,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") job_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1") + scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler") worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") def fake_signed_request( @@ -311,6 +312,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {"job": {"job-id": "job-1"}} if (method, path) == ("GET", job_path): return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}} + if (method, path) == ("GET", scheduler_path): + return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1"}]} if (method, path) == ("POST", worker_path): return {"job": {"status": "UNKNOWN"}} raise AssertionError(f"unexpected REST request: {method} {path}") @@ -330,6 +333,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): view_path = pyiceberg_smoke.view_endpoint_path(args) config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config") maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata") + scheduler_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/scheduler") worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run") diagnostics_path = pyiceberg_smoke.table_endpoint_path(args, "/catalog/diagnostics") @@ -369,6 +373,8 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): return {"job": {"job-id": "job-1"}} if (method, path) == ("GET", pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")): return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}} + if (method, path) == ("GET", scheduler_path): + return {"status": "DISABLED", "audit_timeline": [{"job_id": "job-1"}]} if (method, path) == ("POST", worker_path): return {"job": {"status": "PAUSED"}} if (method, path) == ("GET", diagnostics_path): @@ -398,6 +404,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): ) self.assertIn(("GET", metadata_location_path, None), calls) self.assertIn(("GET", diagnostics_path, None), calls) + self.assertIn(("GET", scheduler_path, None), calls) self.assertIn(("POST", worker_path, {}), calls) def test_table_ref_probe_force_deletes_smoke_ref_after_validation_failure(self) -> None: