feat: complete catalog-driven preprocessing

harness/tht/jobs/dwh_pipeline.py

@@ -42,7 +42,7 @@ def _cleanup_snapshot_dirs() -> None:
 atexit.register(_cleanup_snapshot_dirs)
 
 
-def config_dwh_binding(cfg) -> dict[str, str]:
+def config_dwh_binding(cfg) -> dict[str, str | int]:
     workspace_id = getattr(cfg, "_workspace_id", None)
     config_source = getattr(cfg, "_config_source", None)
     if not isinstance(workspace_id, str) or not isinstance(config_source, str):
@@ -72,14 +72,24 @@ def config_dwh_binding(cfg) -> dict[str, str]:
         else:
             config_fingerprint = fingerprint(cfg.model_dump_json())
             input_fingerprint = fingerprint(config_source)
-    return {
+    binding: dict[str, str | int] = {
         "workspace_id": workspace_id,
         "config_fingerprint": config_fingerprint,
         "input_fingerprint": input_fingerprint,
     }
+    snapshot_path = getattr(getattr(cfg, "paths", None), "catalog_metadata_snapshot", None)
+    if snapshot_path is not None:
+        from tht.mschema.catalog_snapshot import load_catalog_metadata_snapshot
+
+        snapshot = load_catalog_metadata_snapshot(snapshot_path, workspace_id)
+        binding.update({
+            "catalog_database_id": snapshot.database_id,
+            "metadata_content_revision": snapshot.metadata_content_revision,
+        })
+    return binding
 
 
-def _binding_digest(binding: dict[str, str]) -> str:
+def _binding_digest(binding: dict[str, str | int]) -> str:
     payload = json.dumps(binding, sort_keys=True, separators=(",", ":"))
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
@@ -103,10 +113,19 @@ def _read_root_binding_fd(root_fd: int) -> dict[str, str]:
             os.close(fd)
         payload = json.loads(b"".join(chunks).decode("utf-8"))
         binding = payload["binding"]
+        keys = set(binding) if isinstance(binding, dict) else set()
+        base_keys = {"workspace_id", "config_fingerprint", "input_fingerprint"}
+        catalog_keys = {"catalog_database_id", "metadata_content_revision"}
         if (
-            payload.get("schema_version") != 1
+            payload.get("schema_version") not in {1, 2}
             or not isinstance(binding, dict)
-            or set(binding) != {"workspace_id", "config_fingerprint", "input_fingerprint"}
+            or frozenset(keys) not in {frozenset(base_keys), frozenset(base_keys | catalog_keys)}
+            or (keys == base_keys | catalog_keys and (
+                not isinstance(binding["catalog_database_id"], str)
+                or not binding["catalog_database_id"]
+                or not isinstance(binding["metadata_content_revision"], int)
+                or binding["metadata_content_revision"] < 0
+            ))
             or payload.get("binding_sha256") != _binding_digest(binding)
         ):
             raise ValueError
@@ -168,7 +187,7 @@ def _claim_or_validate_root_binding(
             finally:
                 os.close(generations_fd)
         payload = {
-            "schema_version": 1,
+            "schema_version": 2 if "catalog_database_id" in binding else 1,
             "binding": binding,
             "binding_sha256": _binding_digest(binding),
         }
@@ -557,6 +576,8 @@ class DwhPreprocessPipeline:
         workspace_root: Path,
         config_fingerprint: str,
         input_fingerprint: str,
+        catalog_database_id: str | None = None,
+        metadata_content_revision: int | None = None,
         introspect: Callable[[Path], object],
         build_lsh: Callable[[Path, Path], object],
         lsh_filenames: tuple[str, str, str] | None = None,
@@ -569,6 +590,8 @@ class DwhPreprocessPipeline:
         self.workspace_root = workspace_root
         self.config_fingerprint = config_fingerprint
         self.input_fingerprint = input_fingerprint
+        self.catalog_database_id = catalog_database_id
+        self.metadata_content_revision = metadata_content_revision
         self.introspect = introspect
         self.build_lsh = build_lsh
         self.lsh_filenames = lsh_filenames or (
@@ -588,12 +611,20 @@ class DwhPreprocessPipeline:
             raise ValueError("LSH filenames must be unique flat safe names")
 
     @property
-    def binding(self) -> dict[str, str]:
-        return {
+    def binding(self) -> dict[str, str | int]:
+        binding: dict[str, str | int] = {
             "workspace_id": self.workspace_id,
             "config_fingerprint": self.config_fingerprint,
             "input_fingerprint": self.input_fingerprint,
         }
+        if self.catalog_database_id is not None:
+            if self.metadata_content_revision is None:
+                raise ValueError("Catalog metadata revision is required for DWH binding")
+            binding.update({
+                "catalog_database_id": self.catalog_database_id,
+                "metadata_content_revision": self.metadata_content_revision,
+            })
+        return binding
 
     def _assert_active_binding(self, root_fd: int) -> None:
         _validate_root_binding_fd(root_fd, self.binding)