DataOps-platform/app/core/data_rules/compiler.py

"""Deterministic RuleSpec-to-plan compilation for production-line release."""

from __future__ import annotations

import hashlib
import json
from typing import Any

from app.core.common.identifiers import ensure_governance_uid
from app.core.data_rules.contracts import read_rule_spec, rule_spec_hash
from app.core.data_rules.expressions import SUPPORTED_BACKENDS, backend_support


COMPILER_VERSION = "dataops-rulespec-1.0"


def _canonical_hash(value: Any) -> str:
    canonical = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _rule_supported_backends(spec: dict[str, Any]) -> frozenset[str]:
    supported = SUPPORTED_BACKENDS
    for step in spec["steps"]:
        ast = step.get("expression_ast")
        if ast is not None:
            supported = supported & backend_support(ast)
    return frozenset(supported)


def compile_rule_plan(
    rule_version: dict[str, Any], *, supported_backends: frozenset[str] | None = None
) -> dict[str, Any]:
    """Compile a published rule into a target-neutral immutable plan.

    This compiler deliberately emits no SQL or Python. M3 runtime adapters
    translate the closed RuleSpec operators for a deployment's bound backend.
    """

    if not isinstance(rule_version, dict):
        raise ValueError("rule version must be an object")
    if rule_version.get("status") != "published":
        raise ValueError("only published rule versions may be compiled")
    try:
        version_id = ensure_governance_uid(
            {"uid": str(rule_version.get("id"))}
        )
    except ValueError as exc:
        raise ValueError("rule version id must be a valid UUIDv7") from exc
    spec = read_rule_spec(rule_version.get("rule_spec"))
    digest = rule_spec_hash(spec)
    if rule_version.get("spec_hash") != digest:
        raise ValueError("published rule version spec hash does not match")

    derived_backends = _rule_supported_backends(spec)
    if supported_backends is None:
        supported = derived_backends
    elif (
        not isinstance(supported_backends, frozenset)
        or not supported_backends <= SUPPORTED_BACKENDS
    ):
        raise ValueError("supported_backends must be a concrete backend set")
    else:
        supported = supported_backends & derived_backends
    if not supported:
        raise ValueError("rule has no supported execution backend")

    quality_only = all(step["op"] == "assert" for step in spec["steps"])
    if quality_only:
        backend = "quality_check"
    elif "polars" in supported:
        backend = "polars_batch"
    elif supported & {"postgresql", "mysql"}:
        backend = "sql_pushdown"
    else:
        raise ValueError("rule has no supported compiler target")
    plan = {
        "schema_version": "1.0",
        "compiler_version": COMPILER_VERSION,
        "rule_version_id": version_id,
        "rule_spec_hash": digest,
        "input_schema_ref": spec["input_schema_ref"],
        "output_schema_ref": spec["output_schema_ref"],
        "null_policy": spec["null_policy"],
        "timezone": spec["timezone"],
        "supported_execution_backends": sorted(supported),
        "steps": spec["steps"],
    }
    return {
        "backend": backend,
        "compiler_version": COMPILER_VERSION,
        "plan": plan,
        "plan_hash": _canonical_hash(plan),
    }