feat(table-catalog): add durable backing state transfer (#4952)

* feat(table-catalog): add durable backing state transfer

* fix(table-catalog): reject orphaned migration entries

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: Zhengchao An <anzhengchao@gmail.com>

scripts/table-catalog/failure_coverage.py

@@ -335,7 +335,14 @@ def disaster_recovery_rehearsal_plan(
                         "GET",
                         warehouse_path(warehouse, "/catalog/migration", rest_path),
                         "200",
-                        "migration blockers must be empty before cutover",
+                        "migration blockers must be empty and status must be READY_TO_SNAPSHOT before state transfer",
+                    ),
+                    probe_step(
+                        "durable-backing-migration-materialize",
+                        "POST",
+                        warehouse_path(warehouse, "/catalog/migration", rest_path),
+                        "200",
+                        "source writes are fenced and the durable strong snapshot is materialized before cutover",
                     ),
                 ],
             ),
@@ -504,6 +511,13 @@ def scale_fault_rehearsal_plan(
                         "200",
                         "migration dry-run reports inventory, replay blockers, idempotency blockers, and recommended actions",
                     ),
+                    probe_step(
+                        "materialize-durable-backing-before-cutover",
+                        "POST",
+                        warehouse_path(warehouse, "/catalog/migration", rest_path),
+                        "200",
+                        "materialization drains in-flight writes and reports SNAPSHOT_MATERIALIZED before backing restart",
+                    ),
                 ],
             ),
             rehearsal_phase(

scripts/table-catalog/pyiceberg_smoke.py

@@ -239,8 +239,8 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
 PRODUCTION_READINESS_INVENTORY: list[dict[str, str]] = [
     {
         "capability": "strong-catalog-backing",
-        "status": "migration-contract-supported",
-        "validation": "catalog export and diagnostics expose an object-backed manifest, recoverable commit-log WAL state, strong-kv-wal migration target, required migration steps, and recovery blockers before cutover",
+        "status": "state-transfer-supported",
+        "validation": "preflight exposes recovery blockers and target agreement; migration execution fences object-backed writes, drains in-flight catalog mutations, and idempotently materializes table bucket, namespace, table, view, commit-log, and idempotency state before cutover",
     },
     {
         "capability": "single-active-writer-ha",

scripts/table-catalog/test_failure_coverage.py

@@ -132,7 +132,11 @@ class FailureCoverageTest(unittest.TestCase):
 
         migration_steps = {step["name"]: step for step in phases["migration-preflight"]["steps"]}
         self.assertEqual(migration_steps["durable-backing-migration-dry-run"]["path"], "/iceberg/v1/lake/catalog/migration")
-        self.assertEqual(migration_steps["durable-backing-migration-dry-run"]["expected_behavior"], "migration blockers must be empty before cutover")
+        self.assertEqual(migration_steps["durable-backing-migration-materialize"]["method"], "POST")
+        self.assertIn(
+            "READY_TO_SNAPSHOT",
+            migration_steps["durable-backing-migration-dry-run"]["expected_behavior"],
+        )
 
         validation_steps = {step["name"]: step for step in phases["post-recovery-validation"]["steps"]}
         self.assertEqual(validation_steps["load-table-after-recovery"]["path"], "/iceberg/v1/lake/namespaces/sales/tables/orders")
@@ -202,6 +206,7 @@ class FailureCoverageTest(unittest.TestCase):
 
         cutover_steps = {step["name"]: step for step in phases["durable-backing-cutover-under-load"]["steps"]}
         self.assertEqual(cutover_steps["migration-dry-run-before-cutover"]["path"], "/iceberg/v1/lake/catalog/migration")
+        self.assertEqual(cutover_steps["materialize-durable-backing-before-cutover"]["method"], "POST")
         self.assertIn("catalog-backup", cutover_steps["capture-cutover-backup"]["expected_behavior"])
 
         recovery_steps = {step["name"]: step for step in phases["recovery-rollback-import-under-load"]["steps"]}

scripts/table-catalog/test_pyiceberg_smoke.py

@@ -864,7 +864,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
         self.assertIn("single-active-writer-ha", capabilities)
         self.assertIn("scale-validation-matrix", capabilities)
         strong_backing = next(entry for entry in inventory if entry["capability"] == "strong-catalog-backing")
-        self.assertEqual(strong_backing["status"], "migration-contract-supported")
+        self.assertEqual(strong_backing["status"], "state-transfer-supported")
         for entry in inventory:
             self.assertIn("status", entry)
             self.assertIn("validation", entry)