fix: preserve alias in search load checkpoint

backend/search/minjust_opensearch.py

@@ -246,6 +246,7 @@ def checkpoint_state(
     cluster_uuid: str,
     index_uuid: str,
     limit: int | None,
+    alias: str | None,
 ) -> tuple[dict, str | None]:
     source = input_root.resolve()
     destination = checkpoint.resolve()
@@ -254,7 +255,7 @@ def checkpoint_state(
     manifest_sha256 = file_sha256(input_root / "manifest.sqlite3")
     if not resume:
         return {
-            "schema_version": 1,
+            "schema_version": 2,
             "url": url,
             "cluster_uuid": cluster_uuid,
             "index": index,
@@ -262,6 +263,7 @@ def checkpoint_state(
             "input": str(source),
             "manifest_sha256": manifest_sha256,
             "limit": limit,
+            "alias": alias,
             "last_document_code": None,
             "complete": False,
         }, None
@@ -276,13 +278,14 @@ def checkpoint_state(
         "input",
         "manifest_sha256",
         "limit",
+        "alias",
         "last_document_code",
         "complete",
     }
     if not isinstance(state, dict) or set(state) != expected:
         raise ValueError(f"Invalid checkpoint: {checkpoint}")
     if (
-        state["schema_version"] != 1
+        state["schema_version"] != 2
         or state["url"] != url
         or state["cluster_uuid"] != cluster_uuid
         or state["index"] != index
@@ -292,6 +295,8 @@ def checkpoint_state(
         raise ValueError(f"Checkpoint does not match this load: {checkpoint}")
     if state["limit"] != limit:
         raise ValueError(f"Checkpoint limit does not match --limit: {checkpoint}")
+    if state["alias"] != alias:
+        raise ValueError(f"Checkpoint alias does not match --alias: {checkpoint}")
     if state["manifest_sha256"] != manifest_sha256:
         raise ValueError("Normalized manifest changed; create a new versioned index")
     if state["complete"] is not False:
@@ -331,6 +336,7 @@ def load_bulk(
         cluster_uuid,
         index_uuid,
         limit,
+        alias,
     )
     documents = document_count(input_root, limit, start_at)
     if not resume: