feat(table-catalog): add distributed maintenance scheduling (#4257)

* feat(table-catalog): add distributed maintenance scheduling

* fix(table-catalog): address scheduler review feedback

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>

scripts/table-catalog/README.md

@@ -230,9 +230,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, controlled worker
-  execution checks, and a scheduler status report with disabled, paused,
-  backpressure, retry, quarantine, and audit-timeline state; job reports expose
+- metadata maintenance supports safe dry-run planning, controlled scheduler
+  queueing, worker execution checks, and a scheduler status report with disabled,
+  paused, queued, backpressure, retry, quarantine, and audit-timeline state; job reports expose
   structured audit events for planning, worker transitions, heartbeat updates,
   lease expiry, and mutating quarantine operations; quarantined jobs can be
   inspected, released, retried, or abandoned through an operator endpoint
@@ -338,10 +338,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, heartbeat, quarantine operation, and scheduler status endpoints are registered; disabled/paused/backpressure/retry/quarantine/audit-timeline state and per-job audit events are machine-readable; continuous in-process scheduling is not claimed
+- background maintenance worker: controlled scheduler run, worker run-once, heartbeat, quarantine operation, and scheduler status endpoints are registered; disabled/paused/queued/backpressure/retry/quarantine/audit-timeline state and per-job audit events are 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 and scheduler status report; built-in periodic scheduling is not claimed
+- automatic maintenance scheduling: external scheduler hook supported through the scheduler run endpoint, worker run endpoint, and scheduler status report; built-in periodic scheduling is not claimed
 - compaction rewrite: controlled run-once support for partition-local and sort-order-preserving Parquet binpack through metadata maintenance; manifests with position or equality delete files produce machine-readable row-level planning and fail closed before rewrite; built-in periodic scheduling, delete-file rewrite, and row-level compaction execution 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

scripts/table-catalog/pyiceberg_smoke.py

@@ -183,7 +183,7 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
         "status": "controlled-run-once-supported",
         "roadmap_area": "maintenance-worker",
         "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, per-job audit events, lease expiry recovery, heartbeat updates, and operator quarantine inspect/release/retry/abandon actions; built-in periodic scheduling is not claimed",
+        "expected_behavior": "background-enabled maintenance can be queued by the scheduler run endpoint, claimed by the worker run endpoint, and inspected through the scheduler status endpoint with disabled/paused state, queued-job handoff, current-job backpressure, retry deferral, quarantine boundary, audit timeline, per-job audit events, lease expiry recovery, heartbeat updates, and operator quarantine inspect/release/retry/abandon actions; built-in periodic scheduling is not claimed",
     },
     {
         "capability": "manifest-data-reachability-cleanup",
@@ -1050,6 +1050,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")
+    scheduler_run_path = table_endpoint_path(args, "/maintenance/scheduler/run")
     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:
@@ -1083,7 +1084,7 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     if quarantine.get("action") != "INSPECT" or not isinstance(quarantine.get("report"), dict):
         raise RuntimeError("maintenance quarantine endpoint did not return an inspection report")
     scheduler = signed_rest_request(args, deps, "GET", scheduler_path)
-    expected_scheduler_statuses = {"READY", "DISABLED", "PAUSED", "BACKPRESSURED", "RETRY_DEFERRED", "QUARANTINED"}
+    expected_scheduler_statuses = {"READY", "QUEUED", "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")
     scheduler_timeline = scheduler.get("audit_timeline")
@@ -1092,9 +1093,23 @@ def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
     if scheduler_timeline and not isinstance(scheduler_timeline[0].get("audit-events"), list):
         raise RuntimeError("maintenance scheduler audit timeline did not include job audit events")
 
+    scheduler_config = default_maintenance_config()
+    scheduler_config["background-enabled"] = True
+    scheduler_config["worker-paused"] = False
+    signed_rest_request(args, deps, "PUT", config_path, scheduler_config)
+    scheduler_run = signed_rest_request(args, deps, "POST", scheduler_run_path, {"scheduler-id": "pyiceberg-smoke-scheduler"})
+    scheduler_run_job = scheduler_run.get("report", {}).get("job")
+    if not isinstance(scheduler_run_job, dict) or scheduler_run_job.get("status") != "QUEUED":
+        raise RuntimeError("maintenance scheduler run endpoint did not queue a maintenance job")
+    if scheduler_run_job.get("scheduler-id") != "pyiceberg-smoke-scheduler":
+        raise RuntimeError("maintenance scheduler run endpoint did not preserve scheduler identity")
+    scheduler_run_report = scheduler_run.get("scheduler")
+    if not isinstance(scheduler_run_report, dict) or scheduler_run_report.get("status") != "QUEUED":
+        raise RuntimeError("maintenance scheduler run endpoint did not return queued scheduler state")
+
     worker = signed_rest_request(args, deps, "POST", table_endpoint_path(args, "/maintenance/worker/run"), {})
     worker_job = worker.get("job")
-    expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING"}
+    expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING", "QUEUED"}
     if not isinstance(worker_job, dict) or worker_job.get("status") not in expected_statuses:
         raise RuntimeError("maintenance worker endpoint did not return a stable job status")
     if not isinstance(worker.get("audit-events"), list):