feat: execute cross-source Polars rule plans

.superpowers/sdd/task-5-report.md

@@ -0,0 +1,107 @@
+# Task 5 Report — Polars cross-source compilation and artifact I/O
+
+Date: 2026-07-23
+Branch: `codex/data-rule-execution-m3a-m5`
+Lifecycle boundary: physical Polars plans are persisted as `compiled`; this task does not publish them.
+
+## Outcome
+
+Implemented a closed JSON Polars compiler, allowlisted LazyFrame reconstruction, server-owned digest-bound Parquet artifact storage, PostgreSQL-backed artifact binding resolution, Runner registration, canonical plan attestations, and a real PostgreSQL + MySQL + MinIO integration.
+
+The plan contains canonical RuleVersion, SchemaSnapshot, and DatasetBinding hashes plus `dataops-polars-1.42.1` provenance. It contains no Python source, pickle, callable, module, client path, arbitrary URL, secret, or Polars internal serialized plan.
+
+## TDD evidence
+
+RED evidence was captured before each production slice:
+
+- Compiler module: `5 failed` because `app.core.data_rules.compilers.polars` did not exist.
+- ArtifactStore and adapter modules: `8 failed` because the modules did not exist.
+- Cross-source compiler registry/service: `2 failed` because the registry rejected cross-source artifacts.
+- Runner canonical attestation/correlation: `1 failed` because correlation was not forwarded.
+- Node registry trusted context: `1 failed` because `correlation_id` was not accepted.
+- Artifact describe/bootstrap configuration: `2 failed` because the API/settings were absent.
+- Real MinIO integration exposed a production defect: `1 failed` because MinIO returns an `HTTPHeaderDict`, not a plain `dict`, for metadata.
+- Same-size server-side corruption: `1 failed` because write returned before rereading and digest-validating the stored object.
+- Closed operation semantic revalidation: `3 failed` because tampered identifiers/flags were not rejected.
+- Strict MinIO artifact binding contract: `1 failed` because all URI-shaped Parquet refs were previously rejected.
+- Repository-attested binding hashes: `1 failed` because compiler input rejected the canonical `binding_hash`.
+
+GREEN:
+
+- Final required focused command:
+  - `PYTHONPATH=. .venv/bin/pytest -q tests/core/data_rules/test_polars_compiler.py tests/runner/test_rule_polars.py tests/runner/test_artifacts.py tests/integration/test_data_rule_polars_execution.py`
+  - Result: `23 passed in 0.95s`.
+- Expanded Task 5 / SQL regression slice during implementation:
+  - Result: `197 passed`, then `75 passed` after repository hardening.
+
+## Real integration and cleanup
+
+The integration used runtime values discovered from `deploy/docker/docker-compose.yml`; credentials were not copied into the test source.
+
+- PostgreSQL: live source at `127.0.0.1:25432`.
+- MySQL: live lookup source at `127.0.0.1:23306`.
+- MinIO: live object store at `127.0.0.1:19000`.
+- Flow: read PostgreSQL customers, read MySQL segments, write both as bounded Parquet artifacts, compile and execute normalize → lookup join → assert → deduplicate, write and reread the output artifact.
+- Evidence: 4 input rows → 2 output rows, 2 rejected rows, 1 assertion violation; output values and segment enrichment were verified.
+- Cleanup: only `rules/<test-correlation-id>/` objects were removed. The test asserted the exact prefix was empty afterward. Test-owned PostgreSQL/MySQL tables were also dropped.
+
+## Full verification
+
+- `PYTHONPATH=. .venv/bin/pytest -q`
+  - `527 passed, 26 skipped, 59 subtests passed in 4.69s`.
+- `docker compose -f deploy/docker/docker-compose.yml config --quiet`
+  - Passed.
+- Ruff over all Task 5 production and test files
+  - `All checks passed!`
+- `git diff --check`
+  - Passed.
+
+## Dependency and license
+
+- Added exact pin: `polars==1.42.1`.
+- Installed into `.venv`: Polars `1.42.1` and its matching `polars-runtime-32==1.42.1`.
+- Installed metadata: Python `>=3.10`; MIT license text.
+- Plans remain independent of Polars internal serialization formats.
+
+## Files
+
+Created:
+
+- `app/core/data_rules/compilers/polars.py`
+- `app/runner/artifacts.py`
+- `app/runner/rule_polars.py`
+- `tests/core/data_rules/test_polars_compiler.py`
+- `tests/runner/test_artifacts.py`
+- `tests/runner/test_rule_polars.py`
+- `tests/integration/test_data_rule_polars_execution.py`
+
+Modified:
+
+- `requirements.txt`
+- `app/core/data_rules/compilers/__init__.py`
+- `app/core/data_rules/execution_contracts.py`
+- `app/core/data_rules/repository.py`
+- `app/runner/api.py`
+- `app/runner/bootstrap.py`
+- `app/runner/nodes.py`
+- `app/runner/rules.py`
+- `deploy/docker/docker-compose.yml`
+- `tests/core/data_rules/test_execution_contracts.py`
+- `tests/runner/test_bootstrap.py`
+
+## Self-review
+
+- Compiler and runtime validators both enforce closed shapes and operation semantics.
+- Expressions are reconstructed only from the Task 3 AST and must advertise the `polars` backend.
+- Aggregate functions, lookup join variants, masking policy kinds, regexes, casts, assertion action, identifiers, and resource limits are allowlisted.
+- Dataset bindings and lookup contexts are resolved by canonical IDs; repository binding hashes are preserved and re-attested at execution.
+- Artifact keys are generated only by the server as `rules/<correlation-id>/<artifact-id>.parquet`.
+- Artifact reads validate store ownership, strict reference shape, content type, size, TTL, digest, row count, schema digest, and memory bounds.
+- Artifact writes reread the server object before returning a ref, catching same-size corruption.
+- MinIO credentials stay in Runner settings with `repr=False` and never enter refs, logs, or plans.
+
+## Concerns / deferred work
+
+- Task 7 still owns test-evidence recording and plan publication. `RulePlanExecutor` correctly refuses `compiled` plans in the published runtime path.
+- `quality_check` remains fail-closed through the existing adapter; this task did not claim a new artifact-backed quality-check implementation.
+- Digest/schema validation requires reading a bounded artifact before exposing a LazyFrame. Lazy operations and Parquet I/O are used, but fully streaming end-to-end execution is intentionally deferred until a streaming digest-verification design preserves the same security checks.

app/core/data_rules/compilers/__init__.py

@@ -16,7 +16,7 @@ def _dialect(value: Any) -> str:
 
 
 class CompilerRegistry:
-    """Select only a concrete same-source compiler registered by the runtime."""
+    """Select a concrete SQL or artifact-backed batch compiler."""
 
     def __init__(self, compilers: Mapping[str, RuleCompiler]):
         self.compilers = {
@@ -34,6 +34,28 @@ class CompilerRegistry:
             output_binding, dict
         ):
             raise ValueError("dataset bindings must be objects")
+        artifact_batch = (
+            input_binding.get("object_kind") == "parquet_artifact"
+            and output_binding.get("object_kind") == "parquet_artifact"
+        )
+        supported = {"postgresql", "mysql", "polars"}
+        for step in spec["steps"]:
+            ast = step.get("expression_ast")
+            if ast is not None:
+                supported &= set(backend_support(ast))
+        if artifact_batch:
+            if "polars" not in supported:
+                raise ValueError(
+                    "rule semantics are unsupported by the bound batch compiler"
+                )
+            compiler = self.compilers.get("polars")
+            if compiler is None or not callable(
+                getattr(compiler, "compile", None)
+            ):
+                raise ValueError(
+                    "no registered compiler matches the dataset bindings"
+                )
+            return compiler
         if input_binding.get("data_source_uid") != output_binding.get(
             "data_source_uid"
         ):
@@ -41,11 +63,6 @@ class CompilerRegistry:
         dialect = _dialect(input_binding.get("dialect"))
         if dialect != _dialect(output_binding.get("dialect")):
             raise ValueError("input and output binding dialects must match")
-        supported = {"postgresql", "mysql", "polars"}
-        for step in spec["steps"]:
-            ast = step.get("expression_ast")
-            if ast is not None:
-                supported &= set(backend_support(ast))
         if dialect not in supported:
             raise ValueError("rule semantics are unsupported by the bound dialect")
         compiler = self.compilers.get(dialect)

app/core/data_rules/compilers/polars.py

@@ -0,0 +1,894 @@
+"""Compile canonical RuleSpecs into closed, portable Polars batch plans."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from decimal import Decimal, InvalidOperation
+from typing import Any
+
+from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.compilers.base import RuleCompiler
+from app.core.data_rules.contracts import read_rule_spec, rule_spec_hash
+from app.core.data_rules.execution_contracts import (
+    canonical_schema_hash,
+    validate_dataset_binding,
+    validate_schema_snapshot,
+)
+from app.core.data_rules.expressions import (
+    backend_support,
+    type_check_expression,
+    validate_rule_expressions,
+)
+
+COMPILER_VERSION = "dataops-polars-1.42.1"
+PLAN_SCHEMA_VERSION = "1.0"
+RESULT_CONTRACT = {
+    "rows_in": "counted",
+    "rows_out": "counted",
+    "rows_rejected": "counted",
+    "violations": "counted",
+}
+SUPPORTED_OPERATIONS = {
+    "aggregate",
+    "assert",
+    "cast",
+    "deduplicate",
+    "derive",
+    "fill_null",
+    "filter",
+    "lookup_join",
+    "map_values",
+    "mask",
+    "normalize_text",
+    "regex_replace",
+}
+SUPPORTED_AGGREGATES = {"count", "max", "mean", "min", "sum"}
+SUPPORTED_MASK_POLICIES = {"preserve_last_4", "redact"}
+_PLAN_KEYS = {
+    "schema_version",
+    "compiler_version",
+    "rule_version_id",
+    "rule_spec_hash",
+    "input_schema_snapshot_id",
+    "input_schema_hash",
+    "input_fields",
+    "output_schema_snapshot_id",
+    "output_schema_hash",
+    "output_fields",
+    "input_binding_id",
+    "input_binding_hash",
+    "output_binding_id",
+    "output_binding_hash",
+    "resource_limits",
+    "operations",
+    "result_contract",
+}
+_LIMIT_KEYS = {"max_rows", "max_artifact_bytes", "memory_limit_bytes"}
+_FORBIDDEN = re.compile(
+    r"\b(?:generated_python|python|pickle(?:d)?|callable|module)\b"
+    r"|(?:file|https?|s3|gs|ftp)://",
+    re.IGNORECASE,
+)
+_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
+_DIGEST = re.compile(r"^[0-9a-f]{64}$")
+_PORTABLE_REGEX_FORBIDDEN = re.compile(r"\(\?|\\[1-9]|\\[pP]")
+_NUMERIC_TYPES = {"integer", "decimal", "float", "double"}
+_CAST_TYPES = {
+    "boolean",
+    "date",
+    "decimal",
+    "double",
+    "float",
+    "integer",
+    "string",
+    "timestamp",
+    "timestamptz",
+}
+_OPERATION_KEYS = {
+    "aggregate": {"op", "group_by", "aggregations"},
+    "assert": {
+        "op",
+        "expression_ast",
+        "on_failure",
+        "severity",
+        "step_id",
+    },
+    "cast": {"op", "column", "to", "on_error"},
+    "deduplicate": {"op", "keys", "order_by", "keep"},
+    "derive": {"op", "target", "expression_ast"},
+    "fill_null": {"op", "column", "value"},
+    "filter": {"op", "expression_ast"},
+    "lookup_join": {
+        "op",
+        "lookup_binding_id",
+        "lookup_binding_hash",
+        "lookup_schema_snapshot_id",
+        "lookup_schema_hash",
+        "lookup_fields",
+        "left_on",
+        "right_on",
+        "select",
+        "how",
+    },
+    "map_values": {"op", "column", "mapping"},
+    "mask": {"op", "column", "policy_id", "policy_kind"},
+    "normalize_text": {
+        "op",
+        "column",
+        "trim",
+        "lowercase",
+        "uppercase",
+    },
+    "regex_replace": {"op", "column", "pattern", "replacement"},
+}
+
+
+def _canonical_json(value: Any) -> str:
+    try:
+        return json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        )
+    except (TypeError, ValueError) as exc:
+        raise ValueError("bound Polars plan must be JSON serializable") from exc
+
+
+def _hash(value: Any) -> str:
+    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise ValueError(f"{label} must be a valid UUIDv7") from exc
+
+
+def _identifier(value: Any, label: str) -> str:
+    normalized = str(value or "")
+    if _IDENTIFIER.fullmatch(normalized) is None:
+        raise ValueError(f"{label} must be an identifier")
+    return normalized
+
+
+def _snapshot(value: Any, label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    snapshot_id = _uid(value.get("id"), f"{label} id")
+    normalized = validate_schema_snapshot(
+        {key: item for key, item in value.items() if key != "id"}
+    )
+    return {"id": snapshot_id, **normalized}
+
+
+def _binding(value: Any, label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    binding_id = _uid(value.get("id"), f"{label} id")
+    binding_hash = value.get("binding_hash")
+    if binding_hash is not None and _DIGEST.fullmatch(
+        str(binding_hash)
+    ) is None:
+        raise ValueError(f"{label} binding_hash must be a sha256 digest")
+    normalized = validate_dataset_binding(
+        {
+            key: item
+            for key, item in value.items()
+            if key not in {"id", "binding_hash"}
+        }
+    )
+    return {
+        "id": binding_id,
+        **normalized,
+        **({"binding_hash": str(binding_hash)} if binding_hash else {}),
+    }
+
+
+def bound_dataset_binding_hash(value: Any) -> str:
+    """Hash a canonical deployment binding without exposing it in the plan."""
+
+    binding = _binding(value, "dataset binding")
+    return binding.get("binding_hash") or _hash(binding)
+
+
+def _reject_forbidden(value: Any) -> None:
+    if isinstance(value, dict):
+        for key, item in value.items():
+            _reject_forbidden(key)
+            _reject_forbidden(item)
+    elif isinstance(value, list):
+        for item in value:
+            _reject_forbidden(item)
+    elif isinstance(value, str) and _FORBIDDEN.search(value):
+        raise ValueError("bound Polars plan contains forbidden runtime content")
+
+
+def _closed(value: Any, keys: set[str], label: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise ValueError(f"{label} must be an object")
+    if set(value) != keys:
+        raise ValueError(f"{label} must have a closed shape")
+    return copy.deepcopy(value)
+
+
+def _limits(value: Any) -> dict[str, int]:
+    result = _closed(value, _LIMIT_KEYS, "Polars resource limits")
+    for key, minimum, maximum in (
+        ("max_rows", 1, 10_000_000),
+        ("max_artifact_bytes", 1024, 2 * 1024 * 1024 * 1024),
+        ("memory_limit_bytes", 16 * 1024 * 1024, 16 * 1024 * 1024 * 1024),
+    ):
+        item = result[key]
+        if isinstance(item, bool) or not isinstance(item, int):
+            raise ValueError(f"{key} must be an integer")
+        if item < minimum or item > maximum:
+            raise ValueError(f"{key} is outside the supported range")
+    return result
+
+
+def _backend(value: Any) -> dict[str, Any]:
+    keys = {
+        "max_rows",
+        "max_artifact_bytes",
+        "memory_limit_bytes",
+        "masking_policies",
+        "lookup_bindings",
+    }
+    result = _closed(value, keys, "Polars backend capabilities")
+    result["resource_limits"] = _limits(
+        {key: result.pop(key) for key in _LIMIT_KEYS}
+    )
+    policies = result["masking_policies"]
+    if not isinstance(policies, dict) or len(policies) > 100:
+        raise ValueError("masking policies must be a bounded object")
+    normalized_policies = {}
+    for policy_id, policy_kind in policies.items():
+        identifier = _identifier(policy_id, "masking policy id")
+        if policy_kind not in SUPPORTED_MASK_POLICIES:
+            raise ValueError("masking policy kind is unsupported")
+        normalized_policies[identifier] = policy_kind
+    result["masking_policies"] = normalized_policies
+    if not isinstance(result["lookup_bindings"], dict) or len(
+        result["lookup_bindings"]
+    ) > 100:
+        raise ValueError("lookup bindings must be a bounded object")
+    return result
+
+
+def _fields(snapshot: dict[str, Any]) -> dict[str, str]:
+    return {field["name"]: field["type"] for field in snapshot["fields"]}
+
+
+def _field_list(snapshot: dict[str, Any]) -> list[dict[str, Any]]:
+    return copy.deepcopy(snapshot["fields"])
+
+
+def _value_matches_type(value: Any, field_type: str) -> bool:
+    if value is None:
+        return True
+    if field_type == "string":
+        return isinstance(value, str)
+    if field_type == "boolean":
+        return isinstance(value, bool)
+    if field_type == "integer":
+        return isinstance(value, int) and not isinstance(value, bool)
+    if field_type in _NUMERIC_TYPES:
+        if isinstance(value, bool):
+            return False
+        try:
+            Decimal(str(value))
+            return True
+        except (InvalidOperation, TypeError, ValueError):
+            return False
+    if field_type in {"date", "timestamp", "timestamptz"}:
+        return isinstance(value, str)
+    return False
+
+
+def _portable_regex(value: Any) -> str:
+    if not isinstance(value, str) or not value or len(value) > 500:
+        raise ValueError("regex pattern is invalid")
+    if _PORTABLE_REGEX_FORBIDDEN.search(value):
+        raise ValueError("regex pattern is outside the portable subset")
+    try:
+        re.compile(value)
+    except re.error as exc:
+        raise ValueError("regex pattern is invalid") from exc
+    return value
+
+
+def _string_list(
+    value: Any, label: str, *, non_empty: bool = True
+) -> list[str]:
+    if not isinstance(value, list) or len(value) > 100:
+        raise ValueError(f"{label} must be a bounded array")
+    if non_empty and not value:
+        raise ValueError(f"{label} must not be empty")
+    result = [_identifier(item, f"{label} item") for item in value]
+    if len(result) != len(set(result)):
+        raise ValueError(f"{label} must not contain duplicates")
+    return result
+
+
+def _operation(value: Any) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise ValueError("Polars operation must be an object")
+    operation = value.get("op")
+    keys = _OPERATION_KEYS.get(operation)
+    if keys is None:
+        raise ValueError("unsupported Polars operation")
+    result = _closed(value, keys, f"Polars {operation} operation")
+    _reject_forbidden(result)
+    for key in (
+        "column",
+        "target",
+        "policy_id",
+        "step_id",
+    ):
+        if key in result:
+            result[key] = _identifier(result[key], f"{operation} {key}")
+    if operation == "normalize_text":
+        flags = [
+            result["trim"],
+            result["lowercase"],
+            result["uppercase"],
+        ]
+        if not all(isinstance(item, bool) for item in flags):
+            raise ValueError("normalize_text flags must be boolean")
+        if not any(flags) or (result["lowercase"] and result["uppercase"]):
+            raise ValueError("normalize_text operation is invalid")
+    if (
+        operation in {"assert", "derive", "filter"}
+        and "polars" not in backend_support(result["expression_ast"])
+    ):
+        raise ValueError("expression is unsupported by Polars")
+    if operation == "assert" and (
+        result["on_failure"] != "reject"
+        or result["severity"]
+        not in {
+            "info",
+            "warning",
+            "error",
+            "critical",
+        }
+    ):
+        raise ValueError("assert operation is invalid")
+    if operation == "cast" and (
+        result["to"] not in _CAST_TYPES or result["on_error"] != "fail"
+    ):
+        raise ValueError("cast operation is invalid")
+    if operation == "deduplicate":
+        _string_list(result["keys"], "deduplicate keys")
+        _string_list(result["order_by"], "deduplicate order_by")
+        if result["keep"] not in {"first", "last"}:
+            raise ValueError("deduplicate operation is invalid")
+    if operation == "regex_replace":
+        _portable_regex(result["pattern"])
+        if not isinstance(result["replacement"], str):
+            raise ValueError("regex replacement must be a string")
+    if operation == "map_values" and (
+        not isinstance(result["mapping"], dict)
+        or not result["mapping"]
+        or not all(
+            isinstance(key, str) and isinstance(item, str)
+            for key, item in result["mapping"].items()
+        )
+    ):
+        raise ValueError("map_values operation is invalid")
+    if operation == "mask" and result["policy_kind"] not in (
+        SUPPORTED_MASK_POLICIES
+    ):
+        raise ValueError("mask operation is invalid")
+    if operation == "aggregate":
+        _string_list(result["group_by"], "aggregate group_by")
+        if not isinstance(result["aggregations"], list) or not result[
+            "aggregations"
+        ]:
+            raise ValueError("aggregate operation is invalid")
+        for aggregate in result["aggregations"]:
+            if not isinstance(aggregate, dict) or set(aggregate) != {
+                "target",
+                "function",
+                "column",
+            }:
+                raise ValueError("aggregate operation is invalid")
+            _identifier(aggregate["target"], "aggregate target")
+            _identifier(aggregate["column"], "aggregate column")
+            if aggregate["function"] not in SUPPORTED_AGGREGATES:
+                raise ValueError("aggregate operation is invalid")
+    if operation == "lookup_join":
+        for key in (
+            "lookup_binding_hash",
+            "lookup_schema_hash",
+        ):
+            if _DIGEST.fullmatch(str(result[key] or "")) is None:
+                raise ValueError(f"{key} must be a sha256 digest")
+        _uid(result["lookup_binding_id"], "lookup_binding_id")
+        _uid(result["lookup_schema_snapshot_id"], "lookup_schema_snapshot_id")
+        if canonical_schema_hash(result["lookup_fields"]) != result[
+            "lookup_schema_hash"
+        ]:
+            raise ValueError("lookup schema fields do not match their hash")
+        left_on = _string_list(result["left_on"], "lookup left_on")
+        right_on = _string_list(result["right_on"], "lookup right_on")
+        if len(left_on) != len(right_on):
+            raise ValueError("lookup join keys must have equal lengths")
+        if result["how"] not in {"inner", "left"}:
+            raise ValueError("lookup join operation is invalid")
+        if (
+            not isinstance(result["select"], dict)
+            or not result["select"]
+            or not all(
+                _IDENTIFIER.fullmatch(str(key or ""))
+                and _IDENTIFIER.fullmatch(str(item or ""))
+                for key, item in result["select"].items()
+            )
+        ):
+            raise ValueError("lookup select is invalid")
+    return result
+
+
+def validate_bound_polars_plan(value: Any) -> dict[str, Any]:
+    plan = _closed(value, _PLAN_KEYS, "bound Polars plan")
+    if plan["schema_version"] != PLAN_SCHEMA_VERSION:
+        raise ValueError("unsupported bound Polars plan schema version")
+    if plan["compiler_version"] != COMPILER_VERSION:
+        raise ValueError("unsupported bound Polars compiler version")
+    for key in (
+        "rule_version_id",
+        "input_schema_snapshot_id",
+        "output_schema_snapshot_id",
+        "input_binding_id",
+        "output_binding_id",
+    ):
+        plan[key] = _uid(plan[key], key)
+    for key in (
+        "rule_spec_hash",
+        "input_schema_hash",
+        "output_schema_hash",
+        "input_binding_hash",
+        "output_binding_hash",
+    ):
+        if _DIGEST.fullmatch(str(plan[key] or "")) is None:
+            raise ValueError(f"{key} must be a sha256 digest")
+    for prefix in ("input", "output"):
+        fields = plan[f"{prefix}_fields"]
+        if canonical_schema_hash(fields) != plan[f"{prefix}_schema_hash"]:
+            raise ValueError(f"{prefix} schema fields do not match their hash")
+    plan["resource_limits"] = _limits(plan["resource_limits"])
+    raw_operations = plan["operations"]
+    if (
+        not isinstance(raw_operations, list)
+        or not raw_operations
+        or len(raw_operations) > 200
+    ):
+        raise ValueError("Polars operations must be a non-empty bounded array")
+    plan["operations"] = [_operation(item) for item in raw_operations]
+    if plan["result_contract"] != RESULT_CONTRACT:
+        raise ValueError("bound Polars result contract is unsupported")
+    _reject_forbidden(plan)
+    _canonical_json(plan)
+    return plan
+
+
+def bound_polars_plan_hash(value: Any) -> str:
+    return _hash(validate_bound_polars_plan(value))
+
+
+class PolarsRuleCompiler(RuleCompiler):
+    """Compile only allowlisted operators and canonical server-owned context."""
+
+    def compile(
+        self,
+        *,
+        rule_version: dict[str, Any],
+        input_schema: dict[str, Any],
+        output_schema: dict[str, Any],
+        input_binding: dict[str, Any],
+        output_binding: dict[str, Any],
+        backend: dict[str, Any],
+    ) -> dict[str, Any]:
+        if not isinstance(rule_version, dict) or rule_version.get(
+            "status"
+        ) != "published":
+            raise ValueError("only published rule versions may be compiled")
+        rule_version_id = _uid(rule_version.get("id"), "rule_version_id")
+        spec = read_rule_spec(rule_version.get("rule_spec"))
+        if rule_version.get("spec_hash") != rule_spec_hash(spec):
+            raise ValueError("published rule version spec hash does not match")
+        source_schema = _snapshot(input_schema, "input schema")
+        target_schema = _snapshot(output_schema, "output schema")
+        source_binding = _binding(input_binding, "input binding")
+        target_binding = _binding(output_binding, "output binding")
+        capabilities = _backend(backend)
+        if (
+            spec["input_schema_ref"] != source_schema["schema_ref"]
+            or spec["output_schema_ref"] != target_schema["schema_ref"]
+        ):
+            raise ValueError("rule schema references do not match snapshots")
+        if (
+            source_binding["schema_snapshot_id"] != source_schema["id"]
+            or target_binding["schema_snapshot_id"] != target_schema["id"]
+        ):
+            raise ValueError("binding schema snapshot does not match")
+        if source_binding["access_mode"] not in {"read", "read_write"}:
+            raise ValueError("input binding is not readable")
+        if target_binding["access_mode"] not in {"write", "read_write"}:
+            raise ValueError("output binding is not writable")
+        if (
+            source_binding["object_kind"] != "parquet_artifact"
+            or target_binding["object_kind"] != "parquet_artifact"
+        ):
+            raise ValueError("Polars batch rules require Parquet artifact bindings")
+        field_types = _fields(source_schema)
+        supported = validate_rule_expressions(spec, field_types)
+        if "polars" not in supported:
+            raise ValueError("rule expressions are unsupported by polars")
+        output_types = _fields(target_schema)
+        current_fields = list(field_types)
+        operations = []
+
+        for step in spec["steps"]:
+            op = step["op"]
+            if op not in SUPPORTED_OPERATIONS:
+                raise ValueError(f"unsupported Polars rule operation: {op}")
+            if op in {
+                "cast",
+                "fill_null",
+                "map_values",
+                "mask",
+                "normalize_text",
+                "regex_replace",
+            }:
+                column = _identifier(step.get("column"), f"{op} column")
+                if column not in field_types:
+                    raise ValueError(f"unknown Polars rule column: {column}")
+            if op == "normalize_text":
+                if field_types[column] != "string":
+                    raise ValueError("normalize_text requires a string field type")
+                trim = bool(step.get("trim", False))
+                lowercase = bool(step.get("lowercase", False))
+                uppercase = bool(step.get("uppercase", False))
+                if lowercase and uppercase:
+                    raise ValueError("text cannot be lowercased and uppercased")
+                if not (trim or lowercase or uppercase):
+                    raise ValueError("normalize_text has no operation")
+                operations.append(
+                    {
+                        "op": op,
+                        "column": column,
+                        "trim": trim,
+                        "lowercase": lowercase,
+                        "uppercase": uppercase,
+                    }
+                )
+            elif op == "regex_replace":
+                if field_types[column] != "string":
+                    raise ValueError("regex_replace requires a string field type")
+                replacement = step.get("replacement")
+                if not isinstance(replacement, str):
+                    raise ValueError("regex replacement must be a string")
+                operations.append(
+                    {
+                        "op": op,
+                        "column": column,
+                        "pattern": _portable_regex(step.get("pattern")),
+                        "replacement": replacement,
+                    }
+                )
+            elif op == "fill_null":
+                if not _value_matches_type(step.get("value"), field_types[column]):
+                    raise ValueError("fill_null value does not match the field type")
+                operations.append(
+                    {"op": op, "column": column, "value": step.get("value")}
+                )
+            elif op in {"assert", "filter"}:
+                result_type = type_check_expression(
+                    step["expression_ast"], field_types
+                )
+                if result_type != "boolean":
+                    raise ValueError(f"{op} expression must return boolean")
+                operation = {
+                    "op": op,
+                    "expression_ast": copy.deepcopy(step["expression_ast"]),
+                }
+                if op == "assert":
+                    if step.get("on_failure") != "reject":
+                        raise ValueError(
+                            "Polars assert currently supports reject only"
+                        )
+                    operation.update(
+                        {
+                            "on_failure": "reject",
+                            "severity": step.get("severity", "error"),
+                            "step_id": step["id"],
+                        }
+                    )
+                operations.append(operation)
+            elif op == "derive":
+                target = _identifier(step.get("target"), "derive target")
+                if target not in output_types:
+                    raise ValueError("derive target is not in the output schema")
+                result_type = type_check_expression(
+                    step["expression_ast"], field_types
+                )
+                if result_type != output_types[target]:
+                    raise ValueError(
+                        "derive expression type does not match its target type"
+                    )
+                operations.append(
+                    {
+                        "op": op,
+                        "target": target,
+                        "expression_ast": copy.deepcopy(
+                            step["expression_ast"]
+                        ),
+                    }
+                )
+                field_types[target] = result_type
+                if target not in current_fields:
+                    current_fields.append(target)
+            elif op == "map_values":
+                if field_types[column] != "string":
+                    raise ValueError("map_values requires a string field type")
+                mapping = step.get("mapping")
+                if (
+                    not isinstance(mapping, dict)
+                    or not mapping
+                    or len(mapping) > 1_000
+                    or not all(
+                        isinstance(key, str) and isinstance(value, str)
+                        for key, value in mapping.items()
+                    )
+                ):
+                    raise ValueError("map_values requires a string mapping")
+                operations.append(
+                    {
+                        "op": op,
+                        "column": column,
+                        "mapping": dict(sorted(mapping.items())),
+                    }
+                )
+            elif op == "cast":
+                if step.get("on_error", "fail") != "fail":
+                    raise ValueError("Polars cast supports only fail on_error")
+                target_type = str(step.get("to") or "").lower()
+                if target_type not in _CAST_TYPES:
+                    raise ValueError("unsupported Polars cast target")
+                operations.append(
+                    {
+                        "op": op,
+                        "column": column,
+                        "to": target_type,
+                        "on_error": "fail",
+                    }
+                )
+                field_types[column] = target_type
+            elif op == "deduplicate":
+                keys = _string_list(step.get("keys"), "deduplicate keys")
+                order_by = _string_list(
+                    step.get("order_by"), "deduplicate order_by"
+                )
+                if not set(keys + order_by) <= set(current_fields):
+                    raise ValueError("deduplicate references unknown fields")
+                operations.append(
+                    {
+                        "op": op,
+                        "keys": keys,
+                        "order_by": order_by,
+                        "keep": step.get("keep", "first"),
+                    }
+                )
+            elif op == "mask":
+                if field_types[column] != "string":
+                    raise ValueError("mask requires a string field type")
+                policy_id = _identifier(step.get("policy"), "masking policy id")
+                policy_kind = capabilities["masking_policies"].get(policy_id)
+                if policy_kind is None:
+                    raise ValueError("masking policy is not registered")
+                operations.append(
+                    {
+                        "op": op,
+                        "column": column,
+                        "policy_id": policy_id,
+                        "policy_kind": policy_kind,
+                    }
+                )
+            elif op == "aggregate":
+                group_by = _string_list(step.get("group_by"), "aggregate group_by")
+                if not set(group_by) <= set(current_fields):
+                    raise ValueError("aggregate references unknown group fields")
+                raw_aggregations = step.get("aggregations")
+                if (
+                    not isinstance(raw_aggregations, dict)
+                    or not raw_aggregations
+                    or len(raw_aggregations) > 100
+                ):
+                    raise ValueError("aggregations must be a bounded object")
+                aggregations = []
+                next_types = {name: field_types[name] for name in group_by}
+                for target, definition in sorted(raw_aggregations.items()):
+                    target = _identifier(target, "aggregate target")
+                    if not isinstance(definition, dict) or set(definition) != {
+                        "function",
+                        "column",
+                    }:
+                        raise ValueError(
+                            "aggregate definition must have a closed shape"
+                        )
+                    function = definition["function"]
+                    column_name = _identifier(
+                        definition["column"], "aggregate column"
+                    )
+                    if function not in SUPPORTED_AGGREGATES:
+                        raise ValueError("aggregate function is unsupported")
+                    if column_name not in field_types:
+                        raise ValueError("aggregate column is unknown")
+                    result_type = (
+                        "integer"
+                        if function == "count"
+                        else (
+                            "double"
+                            if function == "mean"
+                            else field_types[column_name]
+                        )
+                    )
+                    if output_types.get(target) != result_type:
+                        raise ValueError(
+                            "aggregate result type does not match output schema"
+                        )
+                    next_types[target] = result_type
+                    aggregations.append(
+                        {
+                            "target": target,
+                            "function": function,
+                            "column": column_name,
+                        }
+                    )
+                operations.append(
+                    {
+                        "op": op,
+                        "group_by": group_by,
+                        "aggregations": aggregations,
+                    }
+                )
+                field_types = next_types
+                current_fields = [*group_by, *sorted(raw_aggregations)]
+            elif op == "lookup_join":
+                lookup = step.get("lookup")
+                lookup_keys = {
+                    "binding_id",
+                    "left_on",
+                    "right_on",
+                    "select",
+                    "how",
+                }
+                if not isinstance(lookup, dict) or set(lookup) != lookup_keys:
+                    raise ValueError("lookup_join lookup must have a closed shape")
+                binding_id = _uid(
+                    lookup["binding_id"], "lookup binding id"
+                )
+                context = capabilities["lookup_bindings"].get(binding_id)
+                if not isinstance(context, dict) or set(context) != {
+                    "binding",
+                    "schema",
+                }:
+                    raise ValueError(
+                        "canonical lookup binding was not resolved server side"
+                    )
+                lookup_binding = _binding(
+                    context["binding"], "lookup binding"
+                )
+                lookup_schema = _snapshot(context["schema"], "lookup schema")
+                if (
+                    lookup_binding["id"] != binding_id
+                    or lookup_binding["schema_snapshot_id"]
+                    != lookup_schema["id"]
+                    or lookup_binding["access_mode"]
+                    not in {"read", "read_write"}
+                    or lookup_binding["object_kind"] != "parquet_artifact"
+                ):
+                    raise ValueError("canonical lookup binding is invalid")
+                left_on = _string_list(lookup["left_on"], "lookup left_on")
+                right_on = _string_list(lookup["right_on"], "lookup right_on")
+                if len(left_on) != len(right_on):
+                    raise ValueError("lookup join keys must have equal lengths")
+                lookup_types = _fields(lookup_schema)
+                if (
+                    not set(left_on) <= set(field_types)
+                    or not set(right_on) <= set(lookup_types)
+                    or any(
+                        field_types[left] != lookup_types[right]
+                        for left, right in zip(
+                            left_on, right_on, strict=True
+                        )
+                    )
+                ):
+                    raise ValueError("lookup join key types do not match")
+                select = lookup["select"]
+                if (
+                    not isinstance(select, dict)
+                    or not select
+                    or len(select) > 100
+                ):
+                    raise ValueError("lookup select must be a bounded object")
+                normalized_select = {}
+                for target, source in sorted(select.items()):
+                    target = _identifier(target, "lookup target")
+                    source = _identifier(source, "lookup source")
+                    if source not in lookup_types or target not in output_types:
+                        raise ValueError("lookup select field is unknown")
+                    if lookup_types[source] != output_types[target]:
+                        raise ValueError(
+                            "lookup selected type does not match output schema"
+                        )
+                    normalized_select[target] = source
+                    field_types[target] = lookup_types[source]
+                    if target not in current_fields:
+                        current_fields.append(target)
+                if lookup["how"] not in {"inner", "left"}:
+                    raise ValueError("lookup join kind is unsupported")
+                operations.append(
+                    {
+                        "op": op,
+                        "lookup_binding_id": lookup_binding["id"],
+                        "lookup_binding_hash": (
+                            lookup_binding.get("binding_hash")
+                            or _hash(lookup_binding)
+                        ),
+                        "lookup_schema_snapshot_id": lookup_schema["id"],
+                        "lookup_schema_hash": lookup_schema["schema_hash"],
+                        "lookup_fields": _field_list(lookup_schema),
+                        "left_on": left_on,
+                        "right_on": right_on,
+                        "select": normalized_select,
+                        "how": lookup["how"],
+                    }
+                )
+
+        output_names = [field["name"] for field in target_schema["fields"]]
+        if not set(output_names) <= set(current_fields):
+            raise ValueError("compiled Polars plan cannot produce the output schema")
+        if any(field_types[name] != output_types[name] for name in output_names):
+            raise ValueError(
+                "compiled Polars field types do not match the output schema"
+            )
+        plan = validate_bound_polars_plan(
+            {
+                "schema_version": PLAN_SCHEMA_VERSION,
+                "compiler_version": COMPILER_VERSION,
+                "rule_version_id": rule_version_id,
+                "rule_spec_hash": rule_version["spec_hash"],
+                "input_schema_snapshot_id": source_schema["id"],
+                "input_schema_hash": source_schema["schema_hash"],
+                "input_fields": _field_list(source_schema),
+                "output_schema_snapshot_id": target_schema["id"],
+                "output_schema_hash": target_schema["schema_hash"],
+                "output_fields": _field_list(target_schema),
+                "input_binding_id": source_binding["id"],
+                "input_binding_hash": (
+                    source_binding.get("binding_hash")
+                    or _hash(source_binding)
+                ),
+                "output_binding_id": target_binding["id"],
+                "output_binding_hash": (
+                    target_binding.get("binding_hash")
+                    or _hash(target_binding)
+                ),
+                "resource_limits": capabilities["resource_limits"],
+                "operations": operations,
+                "result_contract": RESULT_CONTRACT,
+            }
+        )
+        return {
+            "backend": "polars_batch",
+            "compiler_version": COMPILER_VERSION,
+            "plan": plan,
+            "plan_hash": bound_polars_plan_hash(plan),
+            "status": "compiled",
+        }

app/core/data_rules/execution_contracts.py

@@ -11,7 +11,6 @@ from typing import Any
 
 from app.core.common.identifiers import ensure_governance_uid
 
-
 SCHEMA_VERSION = "2.0"
 BACKENDS = {"sql_pushdown", "polars_batch", "quality_check"}
 OBJECT_KINDS = {"table", "view", "query", "parquet_artifact"}
@@ -89,6 +88,10 @@ _COMPILER_VERSION_PATTERNS = {
     "quality_check": re.compile(r"^dataops-quality-[1-9][0-9]*(?:\.[0-9]+)+$"),
 }
 _QUERY_REFERENCE = re.compile(r"^query://([0-9a-fA-F-]{36})$")
+_PARQUET_ARTIFACT_REFERENCE = re.compile(
+    r"^minio://([a-z0-9][a-z0-9.-]{1,61}[a-z0-9])/"
+    r"rules/([0-9a-f-]{36})/([0-9a-f-]{36})\.parquet$"
+)
 _PARAMETER_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,99}$")
 _FORBIDDEN_RUNTIME_CONTENT = re.compile(
     r"\b(?:generated_python|python|pickle(?:d)?|callable|module|file)\b"
@@ -249,6 +252,19 @@ def validate_dataset_binding(value: Any) -> dict:
                 "query object_ref must be an opaque query reference"
             )
         binding["object_ref"] = f"query://{_uid(match.group(1), 'query reference')}"
+    elif (
+        binding["object_kind"] == "parquet_artifact"
+        and "://" in binding["object_ref"]
+    ):
+        match = _PARQUET_ARTIFACT_REFERENCE.fullmatch(
+            binding["object_ref"]
+        )
+        if match is None:
+            raise ValueError(
+                "parquet artifact object_ref must be a server-owned MinIO reference"
+            )
+        _uid(match.group(2), "artifact correlation reference")
+        _uid(match.group(3), "artifact reference")
     elif "://" in binding["object_ref"]:
         raise ValueError("dataset object_ref cannot contain a connection string")
     binding["schema_snapshot_id"] = _uid(

app/core/data_rules/repository.py

@@ -929,13 +929,15 @@ class DataRuleRepository:
                     "ib.access_mode AS input_access_mode, "
                     "ib.dialect AS input_dialect, "
                     "ib.write_mode AS input_write_mode, "
+                    "ib.binding_hash AS input_binding_hash, "
                     "ob.id::text AS output_binding_id, "
                     "ob.data_source_uid::text AS output_data_source_uid, "
                     "ob.object_kind AS output_object_kind, "
                     "ob.object_ref AS output_object_ref, "
                     "ob.access_mode AS output_access_mode, "
                     "ob.dialect AS output_dialect, "
-                    "ob.write_mode AS output_write_mode "
+                    "ob.write_mode AS output_write_mode, "
+                    "ob.binding_hash AS output_binding_hash "
                     "FROM public.dataflow_component_bindings cb "
                     "JOIN public.data_rule_versions rv "
                     "ON rv.id = cb.rule_version_id "
@@ -975,24 +977,120 @@ class DataRuleRepository:
                 "access_mode": str(row[f"{prefix}_access_mode"]),
                 "dialect": str(row[f"{prefix}_dialect"]),
                 "write_mode": row[f"{prefix}_write_mode"],
+                "binding_hash": str(row[f"{prefix}_binding_hash"]),
             }
 
         input_snapshot_id = str(row["input_schema_snapshot_id"])
         output_snapshot_id = str(row["output_schema_snapshot_id"])
         rule_spec = _object(row["rule_spec"], "rule_spec")
-        dialect = str(row["input_dialect"]).strip().lower()
-        dialect = "postgresql" if dialect == "postgres" else dialect
-        if dialect not in {"postgresql", "mysql"}:
-            raise ValueError("canonical SQL binding dialect is unsupported")
-        backend = {
-            "dialect": dialect,
-            "timezone": str(rule_spec.get("timezone") or ""),
-            "collation": (
-                "C" if dialect == "postgresql" else "utf8mb4_0900_bin"
-            ),
-            "rounding_mode": "half_away_from_zero",
-            "regex_engine": "posix" if dialect == "postgresql" else "icu",
-        }
+        artifact_batch = (
+            str(row["input_object_kind"]) == "parquet_artifact"
+            and str(row["output_object_kind"]) == "parquet_artifact"
+        )
+        if artifact_batch:
+            backend = {
+                "max_rows": 100_000,
+                "max_artifact_bytes": 32 * 1024 * 1024,
+                "memory_limit_bytes": 256 * 1024 * 1024,
+                "masking_policies": {
+                    "customer_mobile_last4": "preserve_last_4",
+                    "redact": "redact",
+                },
+                "lookup_bindings": {},
+            }
+            for step in rule_spec.get("steps", []):
+                if step.get("op") != "lookup_join":
+                    continue
+                lookup = step.get("lookup")
+                if not isinstance(lookup, dict):
+                    raise ValueError("canonical lookup rule is invalid")
+                lookup_id = _uid(
+                    lookup.get("binding_id"), "lookup binding id"
+                )
+                lookup_row = (
+                    self.session.execute(
+                        text(
+                            "SELECT lb.id::text AS binding_id, "
+                            "lb.data_source_uid::text AS data_source_uid, "
+                            "lb.object_kind, lb.object_ref, "
+                            "lb.schema_snapshot_id::text AS snapshot_id, "
+                            "lb.access_mode, lb.dialect, lb.write_mode, "
+                            "lb.binding_hash, ss.schema_ref, ss.schema_hash, "
+                            "ss.fields, ss.source_revision "
+                            "FROM public.dataflow_dataset_bindings lb "
+                            "JOIN public.data_schema_snapshots ss "
+                            "ON ss.id = lb.schema_snapshot_id "
+                            "JOIN public.dataflow_dataset_bindings ib "
+                            "ON ib.dataflow_deployment_id = "
+                            "lb.dataflow_deployment_id "
+                            "WHERE lb.id = CAST(:lookup_binding_id AS uuid) "
+                            "AND ib.id = CAST(:input_binding_id AS uuid) "
+                            "AND lb.object_kind = 'parquet_artifact' "
+                            "AND lb.access_mode IN ('read', 'read_write') "
+                            "/* canonical_polars_lookup_context */"
+                        ),
+                        {
+                            "lookup_binding_id": lookup_id,
+                            "input_binding_id": ids["input_binding_id"],
+                        },
+                    )
+                    .mappings()
+                    .one_or_none()
+                )
+                if lookup_row is None:
+                    raise ValueError(
+                        "canonical Polars lookup binding was not found"
+                    )
+                snapshot_id = str(lookup_row["snapshot_id"])
+                backend["lookup_bindings"][lookup_id] = {
+                    "binding": {
+                        "id": str(lookup_row["binding_id"]),
+                        "data_source_uid": str(
+                            lookup_row["data_source_uid"]
+                        ),
+                        "object_kind": str(lookup_row["object_kind"]),
+                        "object_ref": str(lookup_row["object_ref"]),
+                        "schema_snapshot_id": snapshot_id,
+                        "access_mode": str(lookup_row["access_mode"]),
+                        "dialect": str(lookup_row["dialect"]),
+                        "write_mode": lookup_row["write_mode"],
+                        "binding_hash": str(
+                            lookup_row["binding_hash"]
+                        ),
+                    },
+                    "schema": {
+                        "id": snapshot_id,
+                        "schema_ref": str(lookup_row["schema_ref"]),
+                        "schema_hash": str(lookup_row["schema_hash"]),
+                        "fields": _array(
+                            lookup_row["fields"],
+                            "lookup schema fields",
+                        ),
+                        "source_revision": str(
+                            lookup_row["source_revision"]
+                        ),
+                    },
+                }
+        else:
+            dialect = str(row["input_dialect"]).strip().lower()
+            dialect = "postgresql" if dialect == "postgres" else dialect
+            if dialect not in {"postgresql", "mysql"}:
+                raise ValueError(
+                    "canonical SQL binding dialect is unsupported"
+                )
+            backend = {
+                "dialect": dialect,
+                "timezone": str(rule_spec.get("timezone") or ""),
+                "collation": (
+                    "C"
+                    if dialect == "postgresql"
+                    else "utf8mb4_0900_bin"
+                ),
+                "rounding_mode": "half_away_from_zero",
+                "regex_engine": (
+                    "posix" if dialect == "postgresql" else "icu"
+                ),
+            }
         return {
             "component_binding": {
                 "id": str(row["component_binding_id"]),
@@ -1039,26 +1137,20 @@ class DataRuleRepository:
         compiled: dict[str, Any],
         status: str = "compiled",
     ) -> dict[str, Any]:
-        """Persist a deployment-bound SQL plan without publishing it.
+        """Persist a deployment-bound SQL or Polars plan without publishing it.
 
         The physical bindings must belong to one deployment whose immutable
         DataFlowVersion owns the logical component binding. Publication and
         test evidence are intentionally separate lifecycle transitions.
         """
 
-        from app.core.data_rules.compilers.sql import (
-            bound_sql_plan_hash,
-            bound_sql_plan_relations,
-            validate_bound_sql_plan,
-        )
-
         component_id = _uid(component_binding_id, "component_binding_id")
         rule_id = _uid(rule_version_id, "rule_version_id")
         input_id = _uid(input_binding_id, "input_binding_id")
         output_id = _uid(output_binding_id, "output_binding_id")
         if status != "compiled":
             raise ValueError(
-                "bound SQL plans may only be persisted as compiled"
+                "bound plans may only be persisted as compiled"
             )
         if not isinstance(compiled, dict) or set(compiled) != {
             "backend",
@@ -1067,23 +1159,42 @@ class DataRuleRepository:
             "plan",
             "plan_hash",
         }:
-            raise ValueError("compiled bound SQL plan has an invalid shape")
-        if compiled["backend"] != "sql_pushdown" or compiled["status"] != status:
-            raise ValueError("compiled bound SQL plan status is invalid")
+            raise ValueError("compiled bound plan has an invalid shape")
+        backend_name = str(compiled["backend"])
+        if (
+            backend_name not in {"sql_pushdown", "polars_batch"}
+            or compiled["status"] != status
+        ):
+            raise ValueError("compiled bound plan status is invalid")
         compiler_version = _text(
             compiled["compiler_version"], "compiler_version", 80
         )
-        plan = validate_bound_sql_plan(compiled["plan"])
+        if backend_name == "sql_pushdown":
+            from app.core.data_rules.compilers.sql import (
+                bound_sql_plan_hash,
+                validate_bound_sql_plan,
+            )
+
+            plan = validate_bound_sql_plan(compiled["plan"])
+            expected_plan_hash = bound_sql_plan_hash(plan)
+        else:
+            from app.core.data_rules.compilers.polars import (
+                bound_polars_plan_hash,
+                validate_bound_polars_plan,
+            )
+
+            plan = validate_bound_polars_plan(compiled["plan"])
+            expected_plan_hash = bound_polars_plan_hash(plan)
         plan_hash = _digest(compiled["plan_hash"], "plan_hash")
-        if bound_sql_plan_hash(plan) != plan_hash:
-            raise ValueError("compiled bound SQL plan hash does not match")
+        if expected_plan_hash != plan_hash:
+            raise ValueError("compiled bound plan hash does not match")
         if (
             plan["rule_version_id"] != rule_id
             or plan["input_binding_id"] != input_id
             or plan["output_binding_id"] != output_id
             or plan["compiler_version"] != compiler_version
         ):
-            raise ValueError("compiled bound SQL plan identifiers do not match")
+            raise ValueError("compiled bound plan identifiers do not match")
 
         linkage = (
             self.session.execute(
@@ -1096,6 +1207,10 @@ class DataRuleRepository:
                     "outs.schema_hash AS output_schema_hash, "
                     "ib.object_ref AS input_object_ref, "
                     "ob.object_ref AS output_object_ref, "
+                    "ib.object_kind AS input_object_kind, "
+                    "ob.object_kind AS output_object_kind, "
+                    "ib.binding_hash AS input_binding_hash, "
+                    "ob.binding_hash AS output_binding_hash, "
                     "ib.data_source_uid::text AS data_source_uid, "
                     "ib.dialect AS input_dialect, "
                     "ob.dialect AS output_dialect "
@@ -1130,25 +1245,49 @@ class DataRuleRepository:
         )
         if linkage is None:
             raise ValueError(
-                "bound SQL plan bindings do not share the component deployment"
+                "bound plan bindings do not share the component deployment"
             )
-        relations = bound_sql_plan_relations(plan)
-        if (
-            str(linkage["data_source_uid"]) != plan["data_source_uid"]
-            or str(linkage["input_object_ref"]) != relations["input_object_ref"]
-            or str(linkage["output_object_ref"]) != relations["output_object_ref"]
-            or str(linkage["input_dialect"]) != plan["dialect"]
-            or str(linkage["output_dialect"]) != plan["dialect"]
-            or str(linkage["rule_spec_hash"]) != plan["rule_spec_hash"]
+        common_mismatch = (
+            str(linkage["rule_spec_hash"]) != plan["rule_spec_hash"]
             or str(linkage["input_schema_snapshot_id"])
             != plan["input_schema_snapshot_id"]
             or str(linkage["input_schema_hash"]) != plan["input_schema_hash"]
             or str(linkage["output_schema_snapshot_id"])
             != plan["output_schema_snapshot_id"]
             or str(linkage["output_schema_hash"]) != plan["output_schema_hash"]
-        ):
+        )
+        if backend_name == "sql_pushdown":
+            from app.core.data_rules.compilers.sql import (
+                bound_sql_plan_relations,
+            )
+
+            relations = bound_sql_plan_relations(plan)
+            mismatch = (
+                common_mismatch
+                or str(linkage["data_source_uid"])
+                != plan["data_source_uid"]
+                or str(linkage["input_object_ref"])
+                != relations["input_object_ref"]
+                or str(linkage["output_object_ref"])
+                != relations["output_object_ref"]
+                or str(linkage["input_dialect"]) != plan["dialect"]
+                or str(linkage["output_dialect"]) != plan["dialect"]
+            )
+        else:
+            mismatch = (
+                common_mismatch
+                or str(linkage["input_object_kind"])
+                != "parquet_artifact"
+                or str(linkage["output_object_kind"])
+                != "parquet_artifact"
+                or str(linkage["input_binding_hash"])
+                != plan["input_binding_hash"]
+                or str(linkage["output_binding_hash"])
+                != plan["output_binding_hash"]
+            )
+        if mismatch:
             raise ValueError(
-                "bound SQL plan does not match its physical dataset bindings"
+                "bound plan does not match its physical dataset bindings"
             )
         row = (
             self.session.execute(
@@ -1157,13 +1296,14 @@ class DataRuleRepository:
                     "(id, component_binding_id, backend, compiler_version, plan, "
                     "plan_hash, schema_hashes, status) "
                     "VALUES (CAST(:id AS uuid), CAST(:component_binding_id AS uuid), "
-                    "'sql_pushdown', :compiler_version, CAST(:plan AS jsonb), "
+                    ":backend, :compiler_version, CAST(:plan AS jsonb), "
                     ":plan_hash, CAST(:schema_hashes AS jsonb), :status) "
                     "RETURNING id::text AS id, status"
                 ),
                 {
                     "id": new_governance_uid(),
                     "component_binding_id": component_id,
+                    "backend": backend_name,
                     "compiler_version": compiler_version,
                     "plan": _json(plan),
                     "plan_hash": plan_hash,
@@ -1189,7 +1329,7 @@ class DataRuleRepository:
         return {
             "id": str(row["id"]),
             "status": str(row["status"]),
-            "backend": "sql_pushdown",
+            "backend": backend_name,
             "plan_hash": plan_hash,
         }
 

app/runner/api.py

@@ -61,6 +61,7 @@ def create_runner_app(*, verifier, ledger, registry):
                 node,
                 parameters,
                 write_authorized=claims.write_authorized,
+                correlation_id=claims.correlation_id,
             )
         except NodeExecutionError as exc:
             recorded = finish_safely(

app/runner/artifacts.py

@@ -0,0 +1,353 @@
+"""Digest-bound, bounded Parquet artifacts owned by the DataOps Runner."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import json
+import os
+import re
+import tempfile
+from collections.abc import Mapping
+from contextlib import suppress
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+import polars as pl
+from sqlalchemy import text
+
+from app.core.common.identifiers import (
+    ensure_governance_uid,
+    new_governance_uid,
+)
+
+PARQUET_CONTENT_TYPE = "application/x-parquet"
+_DIGEST = re.compile(r"^[0-9a-f]{64}$")
+
+
+def _now_utc(clock) -> datetime:
+    value = clock()
+    if not isinstance(value, datetime):
+        raise ValueError("artifact clock must return a datetime")
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)
+
+
+def _timestamp(value: datetime) -> str:
+    return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
+
+
+def _parse_timestamp(value: Any) -> datetime:
+    if not isinstance(value, str) or not value.endswith("Z"):
+        raise ValueError("artifact expiry metadata is invalid")
+    try:
+        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError as exc:
+        raise ValueError("artifact expiry metadata is invalid") from exc
+    return parsed.astimezone(UTC)
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise ValueError(f"{label} must be a valid UUIDv7") from exc
+
+
+def _schema_hash(frame: pl.DataFrame | pl.LazyFrame) -> str:
+    schema = (
+        frame.collect_schema()
+        if isinstance(frame, pl.LazyFrame)
+        else frame.schema
+    )
+    canonical = [
+        {"name": name, "dtype": str(dtype)}
+        for name, dtype in schema.items()
+    ]
+    encoded = json.dumps(
+        canonical,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _metadata(value: Any) -> dict[str, str]:
+    if not isinstance(value, Mapping):
+        raise ValueError("artifact content metadata is missing")
+    normalized = {}
+    for key, item in value.items():
+        name = str(key).lower()
+        if name.startswith("x-amz-meta-"):
+            name = name[len("x-amz-meta-") :]
+        if name in {
+            "sha256",
+            "row-count",
+            "schema-sha256",
+            "expires-at",
+            "artifact-bytes",
+        }:
+            normalized[name] = str(item)
+    required = {
+        "sha256",
+        "row-count",
+        "schema-sha256",
+        "expires-at",
+        "artifact-bytes",
+    }
+    if set(normalized) != required:
+        raise ValueError("artifact content metadata is incomplete")
+    return normalized
+
+
+class ArtifactStore:
+    """Read and write only server-owned, bounded Parquet artifacts."""
+
+    def __init__(
+        self,
+        client,
+        *,
+        bucket: str,
+        max_artifact_bytes: int,
+        max_rows: int,
+        memory_limit_bytes: int,
+        max_ttl_seconds: int = 86400,
+        clock=None,
+    ):
+        if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,61}[a-z0-9]", bucket):
+            raise ValueError("artifact bucket name is invalid")
+        self.client = client
+        self.bucket = bucket
+        self.max_artifact_bytes = int(max_artifact_bytes)
+        self.max_rows = int(max_rows)
+        self.memory_limit_bytes = int(memory_limit_bytes)
+        self.max_ttl_seconds = int(max_ttl_seconds)
+        self.clock = clock or (lambda: datetime.now(UTC))
+        if (
+            self.max_artifact_bytes < 1024
+            or self.max_rows < 1
+            or self.memory_limit_bytes < self.max_artifact_bytes
+            or self.max_ttl_seconds < 1
+        ):
+            raise ValueError("artifact resource limits are invalid")
+        if not self.client.bucket_exists(self.bucket):
+            raise ValueError("artifact bucket does not exist")
+
+    def _parse_ref(self, ref: Any) -> str:
+        prefix = f"minio://{self.bucket}/"
+        if not isinstance(ref, str) or not ref.startswith(prefix):
+            raise ValueError("artifact reference is not owned by this store")
+        key = ref[len(prefix) :]
+        match = re.fullmatch(
+            r"rules/([0-9a-f-]{36})/([0-9a-f-]{36})\.parquet",
+            key,
+        )
+        if match is None:
+            raise ValueError("artifact reference is invalid")
+        _uid(match.group(1), "artifact correlation id")
+        _uid(match.group(2), "artifact id")
+        return key
+
+    def _validated_stat(
+        self,
+        key: str,
+        *,
+        expected_digest: str | None = None,
+    ) -> tuple[Any, dict[str, str]]:
+        stat = self.client.stat_object(self.bucket, key)
+        size = int(getattr(stat, "size", -1))
+        if size < 1 or size > self.max_artifact_bytes:
+            raise ValueError("artifact size exceeds the configured limit")
+        if str(getattr(stat, "content_type", "")).lower() != PARQUET_CONTENT_TYPE:
+            raise ValueError("artifact content type is invalid")
+        metadata = _metadata(getattr(stat, "metadata", None))
+        digest = metadata["sha256"]
+        if _DIGEST.fullmatch(digest) is None:
+            raise ValueError("artifact digest metadata is invalid")
+        if expected_digest is not None and digest != expected_digest:
+            raise ValueError("artifact digest does not match")
+        try:
+            row_count = int(metadata["row-count"])
+            metadata_size = int(metadata["artifact-bytes"])
+        except (TypeError, ValueError) as exc:
+            raise ValueError("artifact count metadata is invalid") from exc
+        if row_count < 0 or row_count > self.max_rows:
+            raise ValueError("artifact row count exceeds the configured limit")
+        if metadata_size != size:
+            raise ValueError("artifact size metadata does not match")
+        if _DIGEST.fullmatch(metadata["schema-sha256"]) is None:
+            raise ValueError("artifact schema metadata is invalid")
+        if _parse_timestamp(metadata["expires-at"]) <= _now_utc(self.clock):
+            raise ValueError("artifact has expired")
+        return stat, metadata
+
+    def write(
+        self,
+        frame: pl.LazyFrame | pl.DataFrame,
+        correlation_id: str,
+        ttl_seconds: int,
+    ) -> dict[str, Any]:
+        correlation = _uid(correlation_id, "correlation_id")
+        if (
+            isinstance(ttl_seconds, bool)
+            or not isinstance(ttl_seconds, int)
+            or ttl_seconds < 1
+            or ttl_seconds > self.max_ttl_seconds
+        ):
+            raise ValueError("artifact TTL is outside the configured limit")
+        if isinstance(frame, pl.DataFrame):
+            lazy = frame.lazy()
+        elif isinstance(frame, pl.LazyFrame):
+            lazy = frame
+        else:
+            raise ValueError("artifact frame must be a Polars frame")
+        collected = lazy.head(self.max_rows + 1).collect(engine="streaming")
+        if collected.height > self.max_rows:
+            raise ValueError("artifact row count exceeds the configured limit")
+        if collected.estimated_size() > self.memory_limit_bytes:
+            raise ValueError("artifact frame exceeds the configured memory limit")
+        schema_digest = _schema_hash(collected)
+        expires_at = _timestamp(
+            _now_utc(self.clock) + timedelta(seconds=ttl_seconds)
+        )
+        artifact_id = new_governance_uid()
+        key = f"rules/{correlation}/{artifact_id}.parquet"
+        path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                prefix="dataops-rule-artifact-",
+                suffix=".parquet",
+                delete=False,
+            ) as handle:
+                path = handle.name
+            collected.write_parquet(path)
+            size = os.path.getsize(path)
+            if size < 1 or size > self.max_artifact_bytes:
+                raise ValueError("artifact size exceeds the configured limit")
+            digest = hashlib.sha256()
+            with open(path, "rb") as handle:
+                while chunk := handle.read(1024 * 1024):
+                    digest.update(chunk)
+            digest_hex = digest.hexdigest()
+            with open(path, "rb") as handle:
+                self.client.put_object(
+                    self.bucket,
+                    key,
+                    handle,
+                    size,
+                    content_type=PARQUET_CONTENT_TYPE,
+                    metadata={
+                        "sha256": digest_hex,
+                        "row-count": str(collected.height),
+                        "schema-sha256": schema_digest,
+                        "expires-at": expires_at,
+                        "artifact-bytes": str(size),
+                    },
+                )
+            self._validated_stat(key, expected_digest=digest_hex)
+            artifact_ref = f"minio://{self.bucket}/{key}"
+            self.read(artifact_ref, digest_hex)
+        finally:
+            if path is not None:
+                with suppress(FileNotFoundError):
+                    os.unlink(path)
+        return {
+            "artifact_ref": artifact_ref,
+            "digest": digest_hex,
+            "row_count": collected.height,
+            "schema_hash": schema_digest,
+            "expires_at": expires_at,
+        }
+
+    def describe(self, ref: str) -> dict[str, Any]:
+        """Return validated object metadata without exposing MinIO credentials."""
+
+        key = self._parse_ref(ref)
+        _stat, metadata = self._validated_stat(key)
+        return {
+            "artifact_ref": ref,
+            "digest": metadata["sha256"],
+            "row_count": int(metadata["row-count"]),
+            "schema_hash": metadata["schema-sha256"],
+            "expires_at": metadata["expires-at"],
+        }
+
+    def read(self, ref: str, expected_digest: str) -> pl.LazyFrame:
+        if _DIGEST.fullmatch(str(expected_digest or "")) is None:
+            raise ValueError("expected artifact digest is invalid")
+        key = self._parse_ref(ref)
+        _stat, metadata = self._validated_stat(
+            key, expected_digest=expected_digest
+        )
+        response = self.client.get_object(self.bucket, key)
+        digest = hashlib.sha256()
+        payload = io.BytesIO()
+        size = 0
+        try:
+            while chunk := response.read(1024 * 1024):
+                size += len(chunk)
+                if size > self.max_artifact_bytes:
+                    raise ValueError(
+                        "artifact size exceeds the configured limit"
+                    )
+                digest.update(chunk)
+                payload.write(chunk)
+        finally:
+            response.close()
+            release = getattr(response, "release_conn", None)
+            if callable(release):
+                release()
+        if digest.hexdigest() != expected_digest:
+            raise ValueError("artifact digest does not match content")
+        payload.seek(0)
+        try:
+            frame = pl.read_parquet(payload)
+        except Exception as exc:
+            raise ValueError("artifact is not valid Parquet") from exc
+        if frame.height != int(metadata["row-count"]):
+            raise ValueError("artifact row count does not match metadata")
+        if frame.height > self.max_rows:
+            raise ValueError("artifact row count exceeds the configured limit")
+        if frame.estimated_size() > self.memory_limit_bytes:
+            raise ValueError("artifact frame exceeds the configured memory limit")
+        if _schema_hash(frame) != metadata["schema-sha256"]:
+            raise ValueError("artifact schema does not match metadata")
+        return frame.lazy()
+
+
+class PostgresArtifactResolver:
+    """Resolve a canonical artifact binding without accepting caller paths."""
+
+    def __init__(self, engine, artifact_store: ArtifactStore):
+        self.engine = engine
+        self.artifact_store = artifact_store
+
+    def resolve(self, *, binding_id: str, correlation_id: str) -> dict[str, Any]:
+        binding = _uid(binding_id, "artifact binding id")
+        correlation = _uid(correlation_id, "artifact correlation id")
+        statement = text(
+            """
+            SELECT object_ref, binding_hash
+            FROM public.dataflow_dataset_bindings
+            WHERE id = CAST(:binding_id AS uuid)
+              AND object_kind = 'parquet_artifact'
+              AND access_mode IN ('read', 'read_write')
+            """
+        )
+        with self.engine.connect() as connection:
+            row = connection.execute(
+                statement, {"binding_id": binding}
+            ).mappings().one_or_none()
+        if row is None:
+            raise ValueError("canonical artifact binding was not found")
+        artifact_ref = str(row["object_ref"])
+        if f"/rules/{correlation}/" not in artifact_ref:
+            raise ValueError(
+                "artifact binding does not match the execution correlation"
+            )
+        return {
+            **self.artifact_store.describe(artifact_ref),
+            "binding_hash": str(row["binding_hash"]),
+        }

app/runner/bootstrap.py

@@ -5,11 +5,14 @@ from __future__ import annotations
 import os
 from dataclasses import dataclass, field
 
+from minio import Minio
+
 from app.core.data_source.runtime import (
     DataSourceRuntimeConfig,
     build_standalone_data_source_runtime,
 )
 from app.runner.api import create_runner_app
+from app.runner.artifacts import ArtifactStore, PostgresArtifactResolver
 from app.runner.auth import TaskTokenVerifier
 from app.runner.ledger import PostgresTaskLedger
 from app.runner.nodes import (
@@ -19,6 +22,7 @@ from app.runner.nodes import (
     SqlExecuteExecutor,
     SqlQueryExecutor,
 )
+from app.runner.rule_polars import PolarsRulePlanAdapter
 from app.runner.rule_sql import (
     SqlGlotQualityPlanAdapter,
     SqlGlotRulePlanAdapter,
@@ -46,10 +50,28 @@ def _integer(name, default, minimum, maximum):
     return value
 
 
+def _boolean(name, default=False):
+    value = str(os.environ.get(name, str(default))).strip().lower()
+    if value in {"1", "true", "yes", "on"}:
+        return True
+    if value in {"0", "false", "no", "off"}:
+        return False
+    raise ValueError(f"{name} must be a boolean")
+
+
 @dataclass(frozen=True)
 class RunnerSettings:
     runtime: DataSourceRuntimeConfig = field(repr=False)
     task_token_secret: str = field(repr=False)
+    artifact_host: str = field(repr=False)
+    artifact_user: str = field(repr=False)
+    artifact_password: str = field(repr=False)
+    artifact_bucket: str = "dataops-rules"
+    artifact_secure: bool = False
+    artifact_max_bytes: int = 32 * 1024 * 1024
+    artifact_max_rows: int = 100_000
+    artifact_memory_limit_bytes: int = 256 * 1024 * 1024
+    artifact_ttl_seconds: int = 3600
     allowed_http_hosts: frozenset = field(default_factory=frozenset)
     task_token_ttl_seconds: int = 60
     max_query_rows: int = 1000
@@ -107,6 +129,29 @@ def runner_settings_from_env():
     return RunnerSettings(
         runtime=runtime,
         task_token_secret=task_token_secret,
+        artifact_host=_required("RUNNER_MINIO_HOST"),
+        artifact_user=_required("RUNNER_MINIO_USER"),
+        artifact_password=_required("RUNNER_MINIO_PASSWORD"),
+        artifact_bucket=_required("RUNNER_MINIO_BUCKET"),
+        artifact_secure=_boolean("RUNNER_MINIO_SECURE", False),
+        artifact_max_bytes=_integer(
+            "RUNNER_ARTIFACT_MAX_BYTES",
+            32 * 1024 * 1024,
+            1024,
+            2 * 1024 * 1024 * 1024,
+        ),
+        artifact_max_rows=_integer(
+            "RUNNER_ARTIFACT_MAX_ROWS", 100_000, 1, 10_000_000
+        ),
+        artifact_memory_limit_bytes=_integer(
+            "RUNNER_ARTIFACT_MEMORY_LIMIT_BYTES",
+            256 * 1024 * 1024,
+            16 * 1024 * 1024,
+            16 * 1024 * 1024 * 1024,
+        ),
+        artifact_ttl_seconds=_integer(
+            "RUNNER_ARTIFACT_TTL_SECONDS", 3600, 1, 86400
+        ),
         allowed_http_hosts=allowed_hosts,
         task_token_ttl_seconds=_integer(
             "RUNNER_TASK_TOKEN_TTL_SECONDS", 60, 1, 300
@@ -124,10 +169,35 @@ def build_runner_application(settings=None):
     )
     write_executor = SqlExecuteExecutor(runtime.manager)
     sql_rule_adapter = SqlGlotRulePlanAdapter(runtime.manager)
+    artifact_store = ArtifactStore(
+        Minio(
+            settings.artifact_host,
+            access_key=settings.artifact_user,
+            secret_key=settings.artifact_password,
+            secure=settings.artifact_secure,
+        ),
+        bucket=settings.artifact_bucket,
+        max_artifact_bytes=settings.artifact_max_bytes,
+        max_rows=settings.artifact_max_rows,
+        memory_limit_bytes=settings.artifact_memory_limit_bytes,
+        max_ttl_seconds=settings.artifact_ttl_seconds,
+    )
+    polars_rule_adapter = PolarsRulePlanAdapter(
+        artifact_store=artifact_store,
+        artifact_resolver=PostgresArtifactResolver(
+            runtime.platform_engine, artifact_store
+        ),
+        masking_policies={
+            "customer_mobile_last4": "preserve_last_4",
+            "redact": "redact",
+        },
+        artifact_ttl_seconds=settings.artifact_ttl_seconds,
+    )
     rule_executor = RulePlanExecutor(
         PostgresRulePlanRepository(runtime.platform_engine),
         adapters={
             "sql_pushdown": sql_rule_adapter,
+            "polars_batch": polars_rule_adapter,
             "quality_check": SqlGlotQualityPlanAdapter(),
         },
     )

app/runner/nodes.py

@@ -7,7 +7,8 @@ import json
 import math
 import multiprocessing
 import re
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 from urllib.parse import urlsplit
 
 import requests
@@ -374,6 +375,7 @@ class NodeRegistry:
         parameters,
         *,
         write_authorized=False,
+        correlation_id=None,
     ):
         executor = self.executors.get(node.get("type"))
         if executor is None or not callable(getattr(executor, "execute", None)):
@@ -382,4 +384,5 @@ class NodeRegistry:
             node,
             parameters,
             write_authorized=write_authorized,
+            correlation_id=correlation_id,
         )

app/runner/rule_polars.py

@@ -0,0 +1,467 @@
+"""Runner adapter that reconstructs allowlisted Polars LazyFrame operations."""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from typing import Any
+
+import polars as pl
+
+from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.compilers.polars import (
+    bound_polars_plan_hash,
+    validate_bound_polars_plan,
+)
+from app.runner.nodes import NodeExecutionError
+
+_TYPE_MAP = {
+    "boolean": pl.Boolean,
+    "date": pl.Date,
+    "double": pl.Float64,
+    "float": pl.Float32,
+    "integer": pl.Int64,
+    "string": pl.String,
+    "timestamp": pl.Datetime,
+    "timestamptz": pl.Datetime,
+}
+_IDEMPOTENCY = {
+    "deduplication_key",
+    "partition_replace",
+    "upsert",
+}
+
+
+def _uid(value: Any, label: str) -> str:
+    try:
+        return ensure_governance_uid({"uid": str(value)})
+    except ValueError as exc:
+        raise NodeExecutionError(f"{label} is invalid") from exc
+
+
+def _dtype_name(dtype: pl.DataType) -> str:
+    if dtype == pl.Boolean:
+        return "boolean"
+    if dtype == pl.Date:
+        return "date"
+    if dtype == pl.String:
+        return "string"
+    if dtype.is_integer():
+        return "integer"
+    if dtype == pl.Float32:
+        return "float"
+    if dtype == pl.Float64:
+        return "double"
+    if dtype.is_decimal():
+        return "decimal"
+    if isinstance(dtype, pl.Datetime):
+        return "timestamptz" if dtype.time_zone else "timestamp"
+    return str(dtype).lower()
+
+
+def _validate_schema(
+    frame: pl.LazyFrame,
+    expected_fields: list[dict[str, Any]],
+    label: str,
+) -> None:
+    schema = frame.collect_schema()
+    expected = {field["name"]: field["type"] for field in expected_fields}
+    actual = {name: _dtype_name(dtype) for name, dtype in schema.items()}
+    if actual != expected:
+        raise NodeExecutionError(
+            f"{label} artifact schema does not match the published plan"
+        )
+
+
+class _ExpressionCompiler:
+    def compile(self, ast: dict[str, Any]) -> pl.Expr:
+        kind = ast["kind"]
+        if kind == "identifier":
+            return pl.col(ast["name"])
+        if kind == "literal":
+            value = ast["value"]
+            if ast["type"] == "decimal":
+                value = Decimal(value)
+            return pl.lit(value)
+        if kind == "unary":
+            operand = self.compile(ast["operand"])
+            return ~operand if ast["operator"] == "!" else -operand
+        if kind == "binary":
+            left = self.compile(ast["left"])
+            right = self.compile(ast["right"])
+            return {
+                "||": lambda: left | right,
+                "&&": lambda: left & right,
+                "==": lambda: left == right,
+                "!=": lambda: left != right,
+                "<": lambda: left < right,
+                "<=": lambda: left <= right,
+                ">": lambda: left > right,
+                ">=": lambda: left >= right,
+                "+": lambda: left + right,
+                "-": lambda: left - right,
+                "*": lambda: left * right,
+                "/": lambda: left / right,
+                "%": lambda: left % right,
+            }[ast["operator"]]()
+        function = ast["function"]
+        arguments = [self.compile(item) for item in ast["arguments"]]
+        if function == "matches":
+            pattern = ast["arguments"][1]["value"]
+            return arguments[0].str.contains(pattern, strict=True)
+        if function == "lower":
+            return arguments[0].str.to_lowercase()
+        if function == "upper":
+            return arguments[0].str.to_uppercase()
+        if function == "trim":
+            return arguments[0].str.strip_chars()
+        if function == "length":
+            return arguments[0].str.len_chars()
+        if function == "coalesce":
+            return pl.coalesce(arguments)
+        if function == "date":
+            return arguments[0].cast(pl.Date, strict=True)
+        if function == "timestamp":
+            return arguments[0].cast(pl.Datetime, strict=True)
+        if function == "abs":
+            return arguments[0].abs()
+        raise NodeExecutionError("published Polars expression is unsupported")
+
+
+def _cast_type(name: str) -> pl.DataType:
+    if name == "decimal":
+        return pl.Decimal(38, 9)
+    dtype = _TYPE_MAP.get(name)
+    if dtype is None:
+        raise NodeExecutionError("published Polars cast type is unsupported")
+    return dtype
+
+
+def _resolve_artifact(
+    resolver,
+    *,
+    binding_id: str,
+    binding_hash: str,
+    correlation_id: str,
+) -> dict[str, Any]:
+    try:
+        artifact = resolver.resolve(
+            binding_id=binding_id,
+            correlation_id=correlation_id,
+        )
+    except Exception as exc:
+        raise NodeExecutionError(
+            "published Polars artifact binding was not resolved"
+        ) from exc
+    if (
+        not isinstance(artifact, dict)
+        or artifact.get("binding_hash") != binding_hash
+        or not isinstance(artifact.get("artifact_ref"), str)
+        or not isinstance(artifact.get("digest"), str)
+    ):
+        raise NodeExecutionError(
+            "published Polars artifact binding does not match"
+        )
+    return artifact
+
+
+class PolarsRulePlanAdapter:
+    """Execute one digest-bound Polars plan and write one bounded artifact."""
+
+    def __init__(
+        self,
+        *,
+        artifact_store,
+        artifact_resolver,
+        masking_policies=None,
+        artifact_ttl_seconds=3600,
+    ):
+        self.artifact_store = artifact_store
+        self.artifact_resolver = artifact_resolver
+        self.masking_policies = dict(masking_policies or {})
+        self.artifact_ttl_seconds = int(artifact_ttl_seconds)
+
+    def execute(
+        self,
+        *,
+        plan,
+        node,
+        parameters,
+        write_authorized,
+        correlation_id=None,
+    ):
+        try:
+            normalized = validate_bound_polars_plan(plan)
+        except ValueError as exc:
+            raise NodeExecutionError(
+                "published Polars rule plan is invalid"
+            ) from exc
+        config = node.get("config") or {}
+        if config.get("execution_plan_hash") != bound_polars_plan_hash(
+            normalized
+        ):
+            raise NodeExecutionError("published Polars rule plan hash does not match")
+        if config.get("rule_version_id") != normalized["rule_version_id"]:
+            raise NodeExecutionError("published Polars rule id does not match")
+        idempotency = node.get("idempotency")
+        if (
+            node.get("type") != "rule.apply"
+            or node.get("purpose") != "write"
+            or not write_authorized
+            or not isinstance(idempotency, dict)
+            or idempotency.get("strategy") not in _IDEMPOTENCY
+            or not str(idempotency.get("key") or "").strip()
+        ):
+            raise NodeExecutionError(
+                "governed write authorization and idempotency are required"
+            )
+        if parameters not in ({}, None):
+            raise NodeExecutionError(
+                "bound Polars rule plans do not accept runtime parameters"
+            )
+        correlation = _uid(correlation_id, "correlation_id")
+        source = _resolve_artifact(
+            self.artifact_resolver,
+            binding_id=normalized["input_binding_id"],
+            binding_hash=normalized["input_binding_hash"],
+            correlation_id=correlation,
+        )
+        try:
+            frame = self.artifact_store.read(
+                source["artifact_ref"], source["digest"]
+            )
+        except ValueError as exc:
+            raise NodeExecutionError(
+                "published Polars input artifact is invalid"
+            ) from exc
+        _validate_schema(frame, normalized["input_fields"], "input")
+        limits = normalized["resource_limits"]
+        initial = frame.head(limits["max_rows"] + 1).collect(engine="streaming")
+        if initial.height > limits["max_rows"]:
+            raise NodeExecutionError(
+                "published Polars input exceeds its row limit"
+            )
+        if initial.estimated_size() > limits["memory_limit_bytes"]:
+            raise NodeExecutionError(
+                "published Polars input exceeds its memory limit"
+            )
+        rows_in = initial.height
+        frame = initial.lazy()
+        expressions = _ExpressionCompiler()
+        violations = []
+
+        for operation in normalized["operations"]:
+            op = operation["op"]
+            if op == "normalize_text":
+                expression = pl.col(operation["column"])
+                if operation["trim"]:
+                    expression = expression.str.strip_chars()
+                if operation["lowercase"]:
+                    expression = expression.str.to_lowercase()
+                if operation["uppercase"]:
+                    expression = expression.str.to_uppercase()
+                frame = frame.with_columns(
+                    expression.alias(operation["column"])
+                )
+            elif op == "regex_replace":
+                frame = frame.with_columns(
+                    pl.col(operation["column"])
+                    .str.replace_all(
+                        operation["pattern"],
+                        operation["replacement"],
+                    )
+                    .alias(operation["column"])
+                )
+            elif op == "fill_null":
+                frame = frame.with_columns(
+                    pl.col(operation["column"])
+                    .fill_null(operation["value"])
+                    .alias(operation["column"])
+                )
+            elif op == "filter":
+                frame = frame.filter(
+                    expressions.compile(operation["expression_ast"]).fill_null(
+                        False
+                    )
+                )
+            elif op == "assert":
+                predicate = expressions.compile(
+                    operation["expression_ast"]
+                ).fill_null(False)
+                invalid = int(
+                    frame.select((~predicate).sum().alias("count"))
+                    .collect(engine="streaming")
+                    .item()
+                    or 0
+                )
+                violations.append(
+                    {"step_id": operation["step_id"], "count": invalid}
+                )
+                frame = frame.filter(predicate)
+            elif op == "derive":
+                frame = frame.with_columns(
+                    expressions.compile(operation["expression_ast"]).alias(
+                        operation["target"]
+                    )
+                )
+            elif op == "map_values":
+                column = pl.col(operation["column"])
+                frame = frame.with_columns(
+                    column.replace_strict(
+                        operation["mapping"],
+                        default=column,
+                    ).alias(operation["column"])
+                )
+            elif op == "cast":
+                frame = frame.with_columns(
+                    pl.col(operation["column"])
+                    .cast(_cast_type(operation["to"]), strict=True)
+                    .alias(operation["column"])
+                )
+            elif op == "deduplicate":
+                order = [
+                    *operation["order_by"],
+                    *sorted(
+                        name
+                        for name in frame.collect_schema().names()
+                        if name not in operation["order_by"]
+                    ),
+                ]
+                descending = operation["keep"] == "last"
+                frame = (
+                    frame.sort(
+                        order,
+                        descending=[descending] * len(order),
+                        nulls_last=True,
+                    )
+                    .unique(
+                        subset=operation["keys"],
+                        keep="first",
+                        maintain_order=True,
+                    )
+                )
+            elif op == "mask":
+                if self.masking_policies.get(
+                    operation["policy_id"]
+                ) != operation["policy_kind"]:
+                    raise NodeExecutionError(
+                        "published masking policy is not registered"
+                    )
+                column = pl.col(operation["column"])
+                if operation["policy_kind"] == "redact":
+                    masked = pl.when(column.is_null()).then(None).otherwise(
+                        pl.lit("***")
+                    )
+                elif operation["policy_kind"] == "preserve_last_4":
+                    masked = pl.when(column.is_null()).then(None).otherwise(
+                        pl.lit("***") + column.str.slice(-4)
+                    )
+                else:
+                    raise NodeExecutionError(
+                        "published masking policy is unsupported"
+                    )
+                frame = frame.with_columns(masked.alias(operation["column"]))
+            elif op == "aggregate":
+                aggregations = []
+                for aggregate in operation["aggregations"]:
+                    expression = pl.col(aggregate["column"])
+                    function = aggregate["function"]
+                    if function == "count":
+                        expression = expression.count()
+                    elif function == "sum":
+                        expression = expression.sum()
+                    elif function == "min":
+                        expression = expression.min()
+                    elif function == "max":
+                        expression = expression.max()
+                    elif function == "mean":
+                        expression = expression.mean()
+                    aggregations.append(
+                        expression.alias(aggregate["target"])
+                    )
+                frame = frame.group_by(
+                    operation["group_by"], maintain_order=True
+                ).agg(aggregations)
+            elif op == "lookup_join":
+                lookup_artifact = _resolve_artifact(
+                    self.artifact_resolver,
+                    binding_id=operation["lookup_binding_id"],
+                    binding_hash=operation["lookup_binding_hash"],
+                    correlation_id=correlation,
+                )
+                try:
+                    lookup = self.artifact_store.read(
+                        lookup_artifact["artifact_ref"],
+                        lookup_artifact["digest"],
+                    )
+                except ValueError as exc:
+                    raise NodeExecutionError(
+                        "published Polars lookup artifact is invalid"
+                    ) from exc
+                _validate_schema(
+                    lookup, operation["lookup_fields"], "lookup"
+                )
+                duplicate_count = (
+                    lookup.group_by(operation["right_on"])
+                    .len()
+                    .filter(pl.col("len") > 1)
+                    .select(pl.len())
+                    .collect(engine="streaming")
+                    .item()
+                )
+                if duplicate_count:
+                    raise NodeExecutionError(
+                        "published Polars lookup keys are not unique"
+                    )
+                selected_sources = list(operation["select"].values())
+                lookup = lookup.select(
+                    list(dict.fromkeys(
+                        [*operation["right_on"], *selected_sources]
+                    ))
+                ).rename(
+                    {
+                        source_name: target_name
+                        for target_name, source_name in operation[
+                            "select"
+                        ].items()
+                        if source_name not in operation["right_on"]
+                    }
+                )
+                frame = frame.join(
+                    lookup,
+                    left_on=operation["left_on"],
+                    right_on=operation["right_on"],
+                    how=operation["how"],
+                )
+
+        output_names = [field["name"] for field in normalized["output_fields"]]
+        frame = frame.select(output_names)
+        _validate_schema(frame, normalized["output_fields"], "output")
+        bounded = frame.head(limits["max_rows"] + 1).collect(engine="streaming")
+        if bounded.height > limits["max_rows"]:
+            raise NodeExecutionError(
+                "published Polars output exceeds its row limit"
+            )
+        if bounded.estimated_size() > limits["memory_limit_bytes"]:
+            raise NodeExecutionError(
+                "published Polars output exceeds its memory limit"
+            )
+        try:
+            artifact = self.artifact_store.write(
+                bounded.lazy(),
+                correlation,
+                self.artifact_ttl_seconds,
+            )
+        except ValueError as exc:
+            raise NodeExecutionError(
+                "published Polars output artifact write failed"
+            ) from exc
+        violation_count = sum(item["count"] for item in violations)
+        return {
+            **artifact,
+            "rows_in": rows_in,
+            "rows_out": bounded.height,
+            "rows_rejected": max(0, rows_in - bounded.height),
+            "violation_count": violation_count,
+            "violations": violations,
+            "commit_outcome": "committed",
+        }

app/runner/rules.py

@@ -11,6 +11,12 @@ from typing import Any
 from sqlalchemy import text
 
 from app.core.common.identifiers import ensure_governance_uid
+from app.core.data_rules.compilers.polars import (
+    COMPILER_VERSION as POLARS_COMPILER_VERSION,
+)
+from app.core.data_rules.compilers.polars import (
+    validate_bound_polars_plan,
+)
 from app.core.data_rules.compilers.sql import (
     COMPILER_VERSION as SQL_COMPILER_VERSION,
 )
@@ -129,7 +135,7 @@ class RulePlanExecutor:
         parameters,
         *,
         write_authorized=False,
-        **_kwargs,
+        **execution_context,
     ):
         if node.get("type") not in {"rule.apply", "quality.check"}:
             raise NodeExecutionError("unsupported governed rule node")
@@ -225,14 +231,56 @@ class RulePlanExecutor:
                 raise NodeExecutionError(
                     "published rule plan canonical attestation does not match"
                 )
+        elif backend == "polars_batch":
+            try:
+                plan = validate_bound_polars_plan(record.get("plan"))
+            except ValueError as exc:
+                raise NodeExecutionError(
+                    "published rule plan is not executable"
+                ) from exc
+            expected_schema_hashes = {
+                "rule_spec_hash": plan["rule_spec_hash"],
+                "input_schema_snapshot_id": plan[
+                    "input_schema_snapshot_id"
+                ],
+                "input_schema_hash": plan["input_schema_hash"],
+                "output_schema_snapshot_id": plan[
+                    "output_schema_snapshot_id"
+                ],
+                "output_schema_hash": plan["output_schema_hash"],
+            }
+            if (
+                record.get("compiler_version") != POLARS_COMPILER_VERSION
+                or plan["compiler_version"] != POLARS_COMPILER_VERSION
+                or record.get("schema_hashes") != expected_schema_hashes
+                or record.get("canonical_rule_spec_hash")
+                != plan["rule_spec_hash"]
+                or record.get("canonical_input_schema_snapshot_id")
+                != plan["input_schema_snapshot_id"]
+                or record.get("canonical_input_schema_hash")
+                != plan["input_schema_hash"]
+                or record.get("canonical_output_schema_snapshot_id")
+                != plan["output_schema_snapshot_id"]
+                or record.get("canonical_output_schema_hash")
+                != plan["output_schema_hash"]
+            ):
+                raise NodeExecutionError(
+                    "published rule plan canonical attestation does not match"
+                )
         adapter = self.adapters.get(backend)
         if adapter is None or not callable(getattr(adapter, "execute", None)):
             raise NodeExecutionError("rule plan backend is not registered")
+        adapter_context = {}
+        if backend == "polars_batch":
+            adapter_context["correlation_id"] = execution_context.get(
+                "correlation_id"
+            )
         result = adapter.execute(
             plan=record["plan"],
             node=node,
             parameters=parameters,
             write_authorized=write_authorized,
+            **adapter_context,
         )
         if not isinstance(result, dict):
             raise NodeExecutionError("rule plan result must be an object")

deploy/docker/docker-compose.yml

@@ -290,6 +290,15 @@ services:
       RUNNER_DATASOURCE_MAX_IDLE_POOLS: 4
       RUNNER_DATASOURCE_CONNECTION_BUDGET: 32
       RUNNER_HTTP_ALLOWED_HOSTS: ${RUNNER_HTTP_ALLOWED_HOSTS:-}
+      RUNNER_MINIO_HOST: minio:9000
+      RUNNER_MINIO_USER: dataops-test
+      RUNNER_MINIO_PASSWORD: dataops-test-password
+      RUNNER_MINIO_BUCKET: dataops-bucket
+      RUNNER_MINIO_SECURE: "false"
+      RUNNER_ARTIFACT_MAX_BYTES: 33554432
+      RUNNER_ARTIFACT_MAX_ROWS: 100000
+      RUNNER_ARTIFACT_MEMORY_LIMIT_BYTES: 268435456
+      RUNNER_ARTIFACT_TTL_SECONDS: 3600
       DATABASE_URL: postgresql://dataops:dataops-test-password@postgres:5432/dataops
       NEO4J_URI: bolt://neo4j:7687
       NEO4J_USER: neo4j
@@ -313,6 +322,8 @@ services:
         condition: service_healthy
       neo4j:
         condition: service_healthy
+      minio-init:
+        condition: service_completed_successfully
     healthcheck:
       test:
         - CMD

requirements.txt

@@ -11,6 +11,7 @@ alembic==1.13.0
 psycopg2-binary==2.9.9
 PyMySQL==1.1.1
 sqlglot==30.13.0
+polars==1.42.1
 neo4j==5.26.0
 argon2-cffi==25.1.0
 PyJWT==2.10.1

tests/core/data_rules/test_execution_contracts.py

@@ -122,6 +122,21 @@ def test_query_dataset_binding_requires_an_opaque_server_published_reference():
     assert validate_dataset_binding(binding)["object_ref"] == binding["object_ref"]
 
 
+def test_parquet_binding_accepts_only_a_server_owned_minio_artifact_shape():
+    binding = dataset_binding()
+    binding["object_kind"] = "parquet_artifact"
+    binding["object_ref"] = (
+        "minio://dataops-rules/rules/"
+        f"{new_governance_uid()}/{new_governance_uid()}.parquet"
+    )
+
+    assert validate_dataset_binding(binding)["object_ref"] == binding["object_ref"]
+
+    binding["object_ref"] = "https://example.test/client-selected.parquet"
+    with pytest.raises(ValueError, match="artifact|connection string"):
+        validate_dataset_binding(binding)
+
+
 @pytest.mark.parametrize(
     ("mutation", "message"),
     [
@@ -232,3 +247,54 @@ def test_execution_plan_rejects_code_serialization_and_url_like_content(
 
     with pytest.raises(ValueError, match="forbidden runtime content"):
         validate_execution_plan_v2(plan)
+
+
+def test_polars_batch_contract_accepts_only_allowlisted_closed_operations():
+    plan = execution_plan()
+    plan.update(
+        {
+            "backend": "polars_batch",
+            "compiler_version": "dataops-polars-1.42.1",
+            "operations": [
+                {
+                    "kind": "polars",
+                    "op": "normalize_text",
+                    "arguments": {
+                        "column": "mobile",
+                        "trim": True,
+                    },
+                }
+            ],
+        }
+    )
+
+    assert validate_execution_plan_v2(plan)["backend"] == "polars_batch"
+
+    plan["operations"][0]["arguments"] = {
+        "callable": "python module payload"
+    }
+    with pytest.raises(ValueError, match="forbidden runtime content"):
+        validate_execution_plan_v2(plan)
+
+
+def test_quality_check_contract_remains_closed_and_assert_only():
+    plan = execution_plan()
+    plan.update(
+        {
+            "backend": "quality_check",
+            "compiler_version": "dataops-quality-1.0",
+            "operations": [
+                {
+                    "kind": "quality",
+                    "check": "assert",
+                    "arguments": {"expression_digest": "a" * 64},
+                }
+            ],
+        }
+    )
+
+    assert validate_execution_plan_v2(plan)["backend"] == "quality_check"
+
+    plan["operations"][0]["check"] = "generated_python"
+    with pytest.raises(ValueError, match="unsupported quality_check"):
+        validate_execution_plan_v2(plan)

tests/core/data_rules/test_polars_compiler.py

@@ -0,0 +1,531 @@
+from __future__ import annotations
+
+import copy
+import json
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_rules.contracts import rule_spec_hash, validate_rule_spec
+from app.core.data_rules.execution_contracts import canonical_schema_hash
+
+
+def _schema(schema_ref, fields):
+    normalized = [
+        {"name": name, "type": field_type, "nullable": nullable}
+        for name, field_type, nullable in fields
+    ]
+    return {
+        "id": new_governance_uid(),
+        "schema_ref": schema_ref,
+        "schema_hash": canonical_schema_hash(normalized),
+        "fields": normalized,
+        "source_revision": "catalog:task5",
+    }
+
+
+def _binding(schema, *, access_mode, source_uid=None, object_ref="artifact"):
+    return {
+        "id": new_governance_uid(),
+        "data_source_uid": source_uid or new_governance_uid(),
+        "object_kind": "parquet_artifact",
+        "object_ref": object_ref,
+        "schema_snapshot_id": schema["id"],
+        "access_mode": access_mode,
+        "dialect": "parquet",
+        "write_mode": "append",
+    }
+
+
+def _published_rule(input_schema, output_schema, steps):
+    spec = validate_rule_spec(
+        {
+            "schema_version": "2.0",
+            "rule_uid": new_governance_uid(),
+            "name": "task5_customer_rule",
+            "input_schema_ref": input_schema["schema_ref"],
+            "output_schema_ref": output_schema["schema_ref"],
+            "steps": steps,
+            "null_policy": "explicit",
+            "timezone": "Asia/Shanghai",
+        }
+    )
+    return {
+        "id": new_governance_uid(),
+        "status": "published",
+        "rule_spec": spec,
+        "spec_hash": rule_spec_hash(spec),
+    }
+
+
+def _backend(**overrides):
+    value = {
+        "max_rows": 10_000,
+        "max_artifact_bytes": 8 * 1024 * 1024,
+        "memory_limit_bytes": 32 * 1024 * 1024,
+        "masking_policies": {
+            "customer_mobile_last4": "preserve_last_4",
+        },
+        "lookup_bindings": {},
+    }
+    value.update(overrides)
+    return value
+
+
+def _compile(steps=None):
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    fields = [
+        ("customer_id", "integer", False),
+        ("name", "string", True),
+        ("mobile", "string", True),
+    ]
+    input_schema = _schema("bd:customer:raw", fields)
+    output_schema = _schema("bd:customer:clean", fields)
+    input_binding = _binding(input_schema, access_mode="read")
+    output_binding = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        steps
+        or [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            },
+            {
+                "id": "valid_mobile",
+                "op": "assert",
+                "expression": "matches(mobile, '^[0-9]{11}$')",
+                "on_failure": "reject",
+                "severity": "error",
+            },
+        ],
+    )
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=_backend(),
+    )
+    return compiled, {
+        "rule": rule,
+        "input_schema": input_schema,
+        "output_schema": output_schema,
+        "input_binding": input_binding,
+        "output_binding": output_binding,
+    }
+
+
+def test_polars_compiler_emits_closed_lazy_operations_and_canonical_hashes():
+    from app.core.data_rules.compilers.polars import (
+        COMPILER_VERSION,
+        bound_polars_plan_hash,
+    )
+
+    compiled, context = _compile()
+
+    assert compiled["backend"] == "polars_batch"
+    assert compiled["compiler_version"] == COMPILER_VERSION
+    assert compiled["status"] == "compiled"
+    assert compiled["plan"]["operations"][0] == {
+        "op": "normalize_text",
+        "column": "name",
+        "trim": True,
+        "lowercase": False,
+        "uppercase": False,
+    }
+    assert compiled["plan"]["rule_spec_hash"] == context["rule"]["spec_hash"]
+    assert compiled["plan"]["input_schema_hash"] == context[
+        "input_schema"
+    ]["schema_hash"]
+    assert compiled["plan"]["input_binding_hash"]
+    assert compiled["plan"]["output_binding_hash"]
+    assert compiled["plan_hash"] == bound_polars_plan_hash(compiled["plan"])
+    encoded = json.dumps(compiled).lower()
+    for forbidden in ("pickle", "python", "callable", "module", "file://"):
+        assert forbidden not in encoded
+
+
+def test_polars_compiler_requires_polars_expression_capability_and_closed_plan():
+    from app.core.data_rules.compilers.polars import (
+        PolarsRuleCompiler,
+        validate_bound_polars_plan,
+    )
+
+    compiled, context = _compile()
+    tampered = copy.deepcopy(compiled["plan"])
+    tampered["operations"][0]["callable"] = "unsafe"
+    with pytest.raises(ValueError, match="closed|unsupported"):
+        validate_bound_polars_plan(tampered)
+
+    context["rule"] = _published_rule(
+        context["input_schema"],
+        context["output_schema"],
+        [
+            {
+                "id": "round_balance",
+                "op": "derive",
+                "target": "customer_id",
+                "expression": "round(customer_id, 2)",
+            }
+        ],
+    )
+    with pytest.raises(ValueError, match="polars"):
+        PolarsRuleCompiler().compile(
+            rule_version=context["rule"],
+            input_schema=context["input_schema"],
+            output_schema=context["output_schema"],
+            input_binding=context["input_binding"],
+            output_binding=context["output_binding"],
+            backend=_backend(),
+        )
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        lambda operation: operation.update({"column": "../unsafe"}),
+        lambda operation: operation.update({"trim": "yes"}),
+        lambda operation: operation.update(
+            {"trim": False, "lowercase": False, "uppercase": False}
+        ),
+    ],
+)
+def test_bound_polars_plan_revalidates_operation_semantics(mutation):
+    from app.core.data_rules.compilers.polars import (
+        validate_bound_polars_plan,
+    )
+
+    compiled, _context = _compile(
+        [
+            {
+                "id": "trim",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    mutation(compiled["plan"]["operations"][0])
+
+    with pytest.raises(ValueError, match="identifier|boolean|operation"):
+        validate_bound_polars_plan(compiled["plan"])
+
+
+def test_polars_compiler_compiles_registered_mask_without_embedding_callable():
+    compiled, _context = _compile(
+        [
+            {
+                "id": "mask_mobile",
+                "op": "mask",
+                "column": "mobile",
+                "policy": "customer_mobile_last4",
+            }
+        ]
+    )
+
+    assert compiled["plan"]["operations"] == [
+        {
+            "op": "mask",
+            "column": "mobile",
+            "policy_id": "customer_mobile_last4",
+            "policy_kind": "preserve_last_4",
+        }
+    ]
+
+    with pytest.raises(ValueError, match="masking policy"):
+        _compile(
+            [
+                {
+                    "id": "mask_mobile",
+                    "op": "mask",
+                    "column": "mobile",
+                    "policy": "unregistered",
+                }
+            ]
+        )
+
+
+def test_polars_compiler_resolves_lookup_binding_server_side_and_emits_ids_only():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw",
+        [
+            ("customer_id", "integer", False),
+            ("segment_code", "string", True),
+        ],
+    )
+    lookup_schema = _schema(
+        "bd:segment:lookup",
+        [
+            ("code", "string", False),
+            ("segment_name", "string", False),
+        ],
+    )
+    output_schema = _schema(
+        "bd:customer:enriched",
+        [
+            ("customer_id", "integer", False),
+            ("segment_code", "string", True),
+            ("segment_name", "string", True),
+        ],
+    )
+    input_binding = _binding(input_schema, access_mode="read")
+    lookup_binding = _binding(
+        lookup_schema,
+        access_mode="read",
+        object_ref="segment-lookup",
+    )
+    output_binding = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "join_segment",
+                "op": "lookup_join",
+                "lookup": {
+                    "binding_id": lookup_binding["id"],
+                    "left_on": ["segment_code"],
+                    "right_on": ["code"],
+                    "select": {"segment_name": "segment_name"},
+                    "how": "left",
+                },
+            }
+        ],
+    )
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=_backend(
+            lookup_bindings={
+                lookup_binding["id"]: {
+                    "binding": lookup_binding,
+                    "schema": lookup_schema,
+                }
+            }
+        ),
+    )
+
+    operation = compiled["plan"]["operations"][0]
+    assert operation["lookup_binding_id"] == lookup_binding["id"]
+    assert operation["lookup_binding_hash"]
+    assert operation["lookup_schema_hash"] == lookup_schema["schema_hash"]
+    encoded = json.dumps(operation)
+    assert "segment-lookup" not in encoded
+    assert "minio://" not in encoded
+
+
+def test_polars_compiler_supports_explicit_aggregate_shape_only():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:sales:raw",
+        [
+            ("region", "string", False),
+            ("amount", "integer", False),
+        ],
+    )
+    output_schema = _schema(
+        "bd:sales:summary",
+        [
+            ("region", "string", False),
+            ("total_amount", "integer", False),
+            ("sale_count", "integer", False),
+        ],
+    )
+    input_binding = _binding(input_schema, access_mode="read")
+    output_binding = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "sum_sales",
+                "op": "aggregate",
+                "group_by": ["region"],
+                "aggregations": {
+                    "sale_count": {"function": "count", "column": "amount"},
+                    "total_amount": {"function": "sum", "column": "amount"},
+                },
+            }
+        ],
+    )
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=_backend(),
+    )
+    assert compiled["plan"]["operations"][0]["aggregations"] == [
+        {"target": "sale_count", "function": "count", "column": "amount"},
+        {"target": "total_amount", "function": "sum", "column": "amount"},
+    ]
+
+    rule["rule_spec"]["steps"][0]["aggregations"]["total_amount"][
+        "function"
+    ] = "custom_callable"
+    rule["spec_hash"] = rule_spec_hash(rule["rule_spec"])
+    with pytest.raises(ValueError, match="aggregate function"):
+        PolarsRuleCompiler().compile(
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=input_binding,
+            output_binding=output_binding,
+            backend=_backend(),
+        )
+
+
+def test_compiler_registry_selects_registered_polars_for_cross_source_artifacts():
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw", [("customer_id", "integer", False)]
+    )
+    output_schema = _schema(
+        "bd:customer:clean", [("customer_id", "integer", False)]
+    )
+    source = _binding(input_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "dedup",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["customer_id"],
+            }
+        ],
+    )
+    compiler = PolarsRuleCompiler()
+
+    assert (
+        CompilerRegistry({"polars": compiler}).select(
+            rule["rule_spec"], source, target
+        )
+        is compiler
+    )
+
+
+def test_bound_plan_service_resolves_polars_context_by_ids_and_stays_compiled():
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+    from app.core.data_rules.release import BoundSqlPlanService
+
+    input_schema = _schema(
+        "bd:customer:raw", [("customer_id", "integer", False)]
+    )
+    output_schema = _schema(
+        "bd:customer:clean", [("customer_id", "integer", False)]
+    )
+    source = _binding(input_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "dedup",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["customer_id"],
+            }
+        ],
+    )
+    component_binding_id = new_governance_uid()
+
+    class Repository:
+        def __init__(self):
+            self.persisted = None
+
+        def load_bound_compile_context(self, **_ids):
+            return {
+                "component_binding": {
+                    "id": component_binding_id,
+                    "rule_version_id": rule["id"],
+                },
+                "rule_version": rule,
+                "input_schema": input_schema,
+                "output_schema": output_schema,
+                "input_binding": source,
+                "output_binding": target,
+                "backend": _backend(),
+            }
+
+        def persist_bound_component_plan(self, **kwargs):
+            self.persisted = kwargs
+            return {
+                "id": new_governance_uid(),
+                "status": kwargs["status"],
+            }
+
+    repository = Repository()
+    result = BoundSqlPlanService(
+        repository,
+        CompilerRegistry({"polars": PolarsRuleCompiler()}),
+    ).compile_and_persist(
+        component_binding_id=component_binding_id,
+        rule_version_id=rule["id"],
+        input_schema_snapshot_id=input_schema["id"],
+        output_schema_snapshot_id=output_schema["id"],
+        input_binding_id=source["id"],
+        output_binding_id=target["id"],
+    )
+
+    assert result["status"] == "compiled"
+    assert repository.persisted["compiled"]["backend"] == "polars_batch"
+    assert repository.persisted["status"] == "compiled"
+
+
+def test_polars_compiler_preserves_repository_attested_binding_hashes():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    input_schema = _schema(
+        "bd:customer:raw", [("customer_id", "integer", False)]
+    )
+    output_schema = _schema(
+        "bd:customer:clean", [("customer_id", "integer", False)]
+    )
+    source = _binding(input_schema, access_mode="read")
+    target = _binding(output_schema, access_mode="write")
+    source["binding_hash"] = "a" * 64
+    target["binding_hash"] = "b" * 64
+    rule = _published_rule(
+        input_schema,
+        output_schema,
+        [
+            {
+                "id": "dedup",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["customer_id"],
+            }
+        ],
+    )
+
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source,
+        output_binding=target,
+        backend=_backend(),
+    )
+
+    assert compiled["plan"]["input_binding_hash"] == "a" * 64
+    assert compiled["plan"]["output_binding_hash"] == "b" * 64

tests/integration/test_data_rule_polars_execution.py

@@ -0,0 +1,382 @@
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import polars as pl
+from minio import Minio
+from sqlalchemy import create_engine, text
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_rules.contracts import rule_spec_hash, validate_rule_spec
+from app.core.data_rules.execution_contracts import canonical_schema_hash
+
+COMPOSE = (
+    Path(__file__).resolve().parents[2]
+    / "deploy"
+    / "docker"
+    / "docker-compose.yml"
+)
+
+
+def _compose_value(pattern):
+    source = COMPOSE.read_text(encoding="utf-8")
+    match = re.search(pattern, source, flags=re.DOTALL)
+    assert match is not None
+    return match.group(1)
+
+
+def _schema(schema_ref, fields):
+    normalized = [
+        {"name": name, "type": field_type, "nullable": nullable}
+        for name, field_type, nullable in fields
+    ]
+    return {
+        "id": new_governance_uid(),
+        "schema_ref": schema_ref,
+        "schema_hash": canonical_schema_hash(normalized),
+        "fields": normalized,
+        "source_revision": "task5:real-cross-source",
+    }
+
+
+def _binding(schema, *, source_uid, access_mode, object_ref):
+    return {
+        "id": new_governance_uid(),
+        "data_source_uid": source_uid,
+        "object_kind": "parquet_artifact",
+        "object_ref": object_ref,
+        "schema_snapshot_id": schema["id"],
+        "access_mode": access_mode,
+        "dialect": "parquet",
+        "write_mode": "append",
+    }
+
+
+class Resolver:
+    def __init__(self, artifacts):
+        self.artifacts = artifacts
+
+    def resolve(self, *, binding_id, correlation_id):
+        artifact = self.artifacts[binding_id]
+        assert f"/rules/{correlation_id}/" in artifact["artifact_ref"]
+        return artifact
+
+
+def test_real_postgres_mysql_minio_polars_cross_source_execution():
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+    from app.runner.artifacts import ArtifactStore
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    source_user = _compose_value(
+        r"source-postgres:.*?POSTGRES_USER:\s*([^\s]+)"
+    )
+    source_password = _compose_value(
+        r"source-postgres:.*?POSTGRES_PASSWORD:\s*([^\s]+)"
+    )
+    postgres_port = _compose_value(r'"(25432):5432"')
+    mysql_port = _compose_value(r'"(23306):3306"')
+    minio_user = _compose_value(r"MINIO_ROOT_USER:\s*([^\s]+)")
+    minio_password = _compose_value(r"MINIO_ROOT_PASSWORD:\s*([^\s]+)")
+    minio_port = _compose_value(r'"(19000):9000"')
+    bucket = _compose_value(r"mc mb --ignore-existing local/([^\s]+)")
+    postgres = create_engine(
+        f"postgresql+psycopg2://{source_user}:{source_password}"
+        f"@127.0.0.1:{postgres_port}/acceptance",
+        pool_pre_ping=True,
+    )
+    mysql = create_engine(
+        f"mysql+pymysql://{source_user}:{source_password}"
+        f"@127.0.0.1:{mysql_port}/acceptance",
+        pool_pre_ping=True,
+    )
+    minio = Minio(
+        f"127.0.0.1:{minio_port}",
+        access_key=minio_user,
+        secret_key=minio_password,
+        secure=False,
+    )
+    store = ArtifactStore(
+        minio,
+        bucket=bucket,
+        max_artifact_bytes=4 * 1024 * 1024,
+        max_rows=1_000,
+        memory_limit_bytes=16 * 1024 * 1024,
+        max_ttl_seconds=3600,
+    )
+    correlation_id = new_governance_uid()
+    prefix = f"rules/{correlation_id}/"
+    customer_table = "task5_polars_customers"
+    segment_table = "task5_polars_segments"
+
+    try:
+        with postgres.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {customer_table}"))
+            connection.execute(
+                text(
+                    f"CREATE TABLE {customer_table} ("
+                    "customer_id BIGINT NOT NULL, "
+                    "name VARCHAR(100), mobile VARCHAR(30), "
+                    "segment_code VARCHAR(20), version_no BIGINT NOT NULL)"
+                )
+            )
+            connection.execute(
+                text(
+                    f"INSERT INTO {customer_table} "
+                    "(customer_id, name, mobile, segment_code, version_no) "
+                    "VALUES "
+                    "(1, ' Alice ', '13800138000', 'A', 1), "
+                    "(1, ' Alice Updated ', '13800138000', 'A', 2), "
+                    "(2, ' Bad ', 'invalid', 'B', 1), "
+                    "(3, ' Carol ', '13900139000', 'C', 1)"
+                )
+            )
+        with mysql.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}"))
+            connection.execute(
+                text(
+                    f"CREATE TABLE {segment_table} ("
+                    "code VARCHAR(20) PRIMARY KEY, "
+                    "segment_name VARCHAR(100) NOT NULL)"
+                )
+            )
+            connection.execute(
+                text(
+                    f"INSERT INTO {segment_table} (code, segment_name) "
+                    "VALUES ('A', 'Gold'), ('B', 'Basic'), ('C', 'Silver')"
+                )
+            )
+        with postgres.connect() as connection:
+            customer_rows = [
+                dict(row)
+                for row in connection.execute(
+                    text(
+                        f"SELECT customer_id, name, mobile, "
+                        f"segment_code, version_no FROM {customer_table}"
+                    )
+                ).mappings()
+            ]
+        with mysql.connect() as connection:
+            segment_rows = [
+                dict(row)
+                for row in connection.execute(
+                    text(
+                        f"SELECT code, segment_name FROM {segment_table}"
+                    )
+                ).mappings()
+            ]
+        customer_artifact = store.write(
+            pl.DataFrame(customer_rows).lazy(), correlation_id, 900
+        )
+        segment_artifact = store.write(
+            pl.DataFrame(segment_rows).lazy(), correlation_id, 900
+        )
+
+        input_schema = _schema(
+            "bd:task5:customer:raw",
+            [
+                ("customer_id", "integer", False),
+                ("name", "string", True),
+                ("mobile", "string", True),
+                ("segment_code", "string", True),
+                ("version_no", "integer", False),
+            ],
+        )
+        lookup_schema = _schema(
+            "bd:task5:segment:lookup",
+            [
+                ("code", "string", False),
+                ("segment_name", "string", False),
+            ],
+        )
+        output_schema = _schema(
+            "bd:task5:customer:enriched",
+            [
+                ("customer_id", "integer", False),
+                ("name", "string", True),
+                ("mobile", "string", True),
+                ("segment_code", "string", True),
+                ("version_no", "integer", False),
+                ("segment_name", "string", True),
+            ],
+        )
+        input_binding = _binding(
+            input_schema,
+            source_uid=new_governance_uid(),
+            access_mode="read",
+            object_ref="postgres-customer-artifact",
+        )
+        lookup_binding = _binding(
+            lookup_schema,
+            source_uid=new_governance_uid(),
+            access_mode="read",
+            object_ref="mysql-segment-artifact",
+        )
+        output_binding = _binding(
+            output_schema,
+            source_uid=new_governance_uid(),
+            access_mode="write",
+            object_ref="polars-output-artifact",
+        )
+        spec = validate_rule_spec(
+            {
+                "schema_version": "2.0",
+                "rule_uid": new_governance_uid(),
+                "name": "task5_real_cross_source",
+                "input_schema_ref": input_schema["schema_ref"],
+                "output_schema_ref": output_schema["schema_ref"],
+                "steps": [
+                    {
+                        "id": "normalize_name",
+                        "op": "normalize_text",
+                        "column": "name",
+                        "trim": True,
+                    },
+                    {
+                        "id": "join_segment",
+                        "op": "lookup_join",
+                        "lookup": {
+                            "binding_id": lookup_binding["id"],
+                            "left_on": ["segment_code"],
+                            "right_on": ["code"],
+                            "select": {
+                                "segment_name": "segment_name"
+                            },
+                            "how": "left",
+                        },
+                    },
+                    {
+                        "id": "valid_mobile",
+                        "op": "assert",
+                        "expression": "matches(mobile, '^[0-9]{11}$')",
+                        "on_failure": "reject",
+                        "severity": "error",
+                    },
+                    {
+                        "id": "latest_customer",
+                        "op": "deduplicate",
+                        "keys": ["customer_id"],
+                        "order_by": ["version_no"],
+                        "keep": "last",
+                    },
+                ],
+                "null_policy": "explicit",
+                "timezone": "Asia/Shanghai",
+            }
+        )
+        rule = {
+            "id": new_governance_uid(),
+            "status": "published",
+            "rule_spec": spec,
+            "spec_hash": rule_spec_hash(spec),
+        }
+        compiled = PolarsRuleCompiler().compile(
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=input_binding,
+            output_binding=output_binding,
+            backend={
+                "max_rows": 1_000,
+                "max_artifact_bytes": 4 * 1024 * 1024,
+                "memory_limit_bytes": 16 * 1024 * 1024,
+                "masking_policies": {},
+                "lookup_bindings": {
+                    lookup_binding["id"]: {
+                        "binding": lookup_binding,
+                        "schema": lookup_schema,
+                    }
+                },
+            },
+        )
+        lookup_operation = compiled["plan"]["operations"][1]
+        resolver = Resolver(
+            {
+                input_binding["id"]: {
+                    **customer_artifact,
+                    "binding_hash": compiled["plan"][
+                        "input_binding_hash"
+                    ],
+                },
+                lookup_binding["id"]: {
+                    **segment_artifact,
+                    "binding_hash": lookup_operation[
+                        "lookup_binding_hash"
+                    ],
+                },
+            }
+        )
+        node = {
+            "id": "task5_real_polars",
+            "type": "rule.apply",
+            "purpose": "write",
+            "idempotency": {
+                "strategy": "deduplication_key",
+                "key": "customer_id",
+            },
+            "config": {
+                "component_binding_id": new_governance_uid(),
+                "rule_version_id": rule["id"],
+                "execution_plan_hash": compiled["plan_hash"],
+            },
+        }
+        result = PolarsRulePlanAdapter(
+            artifact_store=store,
+            artifact_resolver=resolver,
+            artifact_ttl_seconds=900,
+        ).execute(
+            plan=compiled["plan"],
+            node=node,
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+        assert result["rows_in"] == 4
+        assert result["rows_out"] == 2
+        assert result["rows_rejected"] == 2
+        assert result["violation_count"] == 1
+        assert result["violations"] == [
+            {"step_id": "valid_mobile", "count": 1}
+        ]
+        output = store.read(
+            result["artifact_ref"], result["digest"]
+        ).collect()
+        assert output.sort("customer_id").to_dicts() == [
+            {
+                "customer_id": 1,
+                "mobile": "13800138000",
+                "name": "Alice Updated",
+                "segment_code": "A",
+                "segment_name": "Gold",
+                "version_no": 2,
+            },
+            {
+                "customer_id": 3,
+                "mobile": "13900139000",
+                "name": "Carol",
+                "segment_code": "C",
+                "segment_name": "Silver",
+                "version_no": 1,
+            },
+        ]
+        assert all(
+            item.object_name.startswith(prefix)
+            for item in minio.list_objects(
+                bucket, prefix=prefix, recursive=True
+            )
+        )
+    finally:
+        for item in list(
+            minio.list_objects(bucket, prefix=prefix, recursive=True)
+        ):
+            minio.remove_object(bucket, item.object_name)
+        assert list(
+            minio.list_objects(bucket, prefix=prefix, recursive=True)
+        ) == []
+        with postgres.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {customer_table}"))
+        with mysql.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {segment_table}"))
+        postgres.dispose()
+        mysql.dispose()

tests/runner/test_artifacts.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+import io
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import polars as pl
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+
+
+class Response(io.BytesIO):
+    def release_conn(self):
+        return None
+
+
+class FakeMinio:
+    def __init__(self):
+        self.buckets = {"dataops-rules"}
+        self.objects = {}
+
+    def bucket_exists(self, bucket):
+        return bucket in self.buckets
+
+    def make_bucket(self, bucket):
+        self.buckets.add(bucket)
+
+    def put_object(
+        self,
+        bucket,
+        key,
+        data,
+        length,
+        *,
+        content_type,
+        metadata,
+    ):
+        payload = data.read(length)
+        self.objects[(bucket, key)] = {
+            "payload": payload,
+            "content_type": content_type,
+            "metadata": {
+                f"x-amz-meta-{name.lower()}": str(value)
+                for name, value in metadata.items()
+            },
+        }
+
+    def stat_object(self, bucket, key):
+        item = self.objects[(bucket, key)]
+        return SimpleNamespace(
+            size=len(item["payload"]),
+            content_type=item["content_type"],
+            metadata=item["metadata"],
+        )
+
+    def get_object(self, bucket, key):
+        return Response(self.objects[(bucket, key)]["payload"])
+
+
+def _store(client, *, clock=None, max_rows=100):
+    from app.runner.artifacts import ArtifactStore
+
+    return ArtifactStore(
+        client,
+        bucket="dataops-rules",
+        max_artifact_bytes=1024 * 1024,
+        max_rows=max_rows,
+        memory_limit_bytes=4 * 1024 * 1024,
+        max_ttl_seconds=3600,
+        clock=clock,
+    )
+
+
+def test_artifact_store_generates_key_and_round_trips_digest_bound_lazyframe():
+    client = FakeMinio()
+    store = _store(client)
+    correlation_id = new_governance_uid()
+
+    artifact = store.write(
+        pl.DataFrame(
+            {
+                "customer_id": [1, 2],
+                "name": ["Alice", "Bob"],
+            }
+        ).lazy(),
+        correlation_id,
+        300,
+    )
+
+    assert artifact["artifact_ref"].startswith(
+        f"minio://dataops-rules/rules/{correlation_id}/"
+    )
+    assert artifact["artifact_ref"].endswith(".parquet")
+    assert artifact["digest"]
+    assert artifact["row_count"] == 2
+    assert artifact["schema_hash"]
+    assert artifact["expires_at"].endswith("Z")
+    assert "dataops-test" not in repr(artifact)
+    assert store.describe(artifact["artifact_ref"]) == artifact
+    frame = store.read(artifact["artifact_ref"], artifact["digest"])
+    assert isinstance(frame, pl.LazyFrame)
+    assert frame.collect().to_dicts() == [
+        {"customer_id": 1, "name": "Alice"},
+        {"customer_id": 2, "name": "Bob"},
+    ]
+
+
+@pytest.mark.parametrize(
+    ("mutation", "message"),
+    [
+        (
+            lambda item: item["metadata"].update(
+                {"x-amz-meta-sha256": "0" * 64}
+            ),
+            "digest",
+        ),
+        (
+            lambda item: item["metadata"].update(
+                {"x-amz-meta-schema-sha256": "0" * 64}
+            ),
+            "schema",
+        ),
+        (
+            lambda item: item["metadata"].update(
+                {"x-amz-meta-expires-at": "2000-01-01T00:00:00Z"}
+            ),
+            "expired",
+        ),
+        (
+            lambda item: item.update({"content_type": "text/plain"}),
+            "content type",
+        ),
+    ],
+)
+def test_artifact_store_rejects_tampered_digest_schema_ttl_and_content(
+    mutation, message
+):
+    client = FakeMinio()
+    now = datetime(2026, 7, 23, 10, 0, tzinfo=UTC)
+    store = _store(client, clock=lambda: now)
+    artifact = store.write(
+        pl.DataFrame({"id": [1]}).lazy(),
+        new_governance_uid(),
+        300,
+    )
+    key = artifact["artifact_ref"].split("/", 3)[-1]
+    mutation(client.objects[("dataops-rules", key)])
+
+    with pytest.raises(ValueError, match=message):
+        store.read(artifact["artifact_ref"], artifact["digest"])
+
+
+def test_artifact_store_rejects_rows_size_ttl_and_unowned_references():
+    client = FakeMinio()
+    store = _store(client, max_rows=2)
+
+    with pytest.raises(ValueError, match="row"):
+        store.write(
+            pl.DataFrame({"id": [1, 2, 3]}).lazy(),
+            new_governance_uid(),
+            60,
+        )
+    with pytest.raises(ValueError, match="TTL"):
+        store.write(
+            pl.DataFrame({"id": [1]}).lazy(),
+            new_governance_uid(),
+            7200,
+        )
+    with pytest.raises(ValueError, match="artifact reference"):
+        store.read(
+            "minio://other-bucket/rules/unsafe/value.parquet",
+            "0" * 64,
+        )
+
+
+def test_artifact_store_does_not_return_ref_for_corrupted_server_content():
+    class CorruptingMinio(FakeMinio):
+        def put_object(self, bucket, key, *args, **kwargs):
+            super().put_object(bucket, key, *args, **kwargs)
+            payload = self.objects[(bucket, key)]["payload"]
+            self.objects[(bucket, key)]["payload"] = bytes(
+                [payload[0] ^ 1]
+            ) + payload[1:]
+
+    store = _store(CorruptingMinio())
+
+    with pytest.raises(ValueError, match="digest|size"):
+        store.write(
+            pl.DataFrame({"id": [1]}).lazy(),
+            new_governance_uid(),
+            60,
+        )

tests/runner/test_bootstrap.py

@@ -17,6 +17,10 @@ def test_runner_settings_require_signing_secret_and_explicit_dependencies(
         "DATASOURCE_CREDENTIAL_KEY_VERSION": "v1",
         "RUNNER_TASK_TOKEN_SECRET": "x" * 32,
         "RUNNER_HTTP_ALLOWED_HOSTS": "events.internal, audit.internal",
+        "RUNNER_MINIO_HOST": "minio:9000",
+        "RUNNER_MINIO_USER": "runner-test",
+        "RUNNER_MINIO_PASSWORD": "runner-test-password",
+        "RUNNER_MINIO_BUCKET": "dataops-rules",
     }
     for key, value in required.items():
         monkeypatch.setenv(key, value)
@@ -30,8 +34,10 @@ def test_runner_settings_require_signing_secret_and_explicit_dependencies(
         "events.internal",
         "audit.internal",
     }
+    assert settings.artifact_bucket == "dataops-rules"
     assert "postgresql://platform" not in repr(settings)
     assert required["RUNNER_TASK_TOKEN_SECRET"] not in repr(settings)
+    assert required["RUNNER_MINIO_PASSWORD"] not in repr(settings)
 
 
 def test_runner_refuses_to_start_without_a_task_token_secret(monkeypatch):

tests/runner/test_rule_polars.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+import copy
+
+import polars as pl
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.runner.nodes import NodeExecutionError
+from tests.core.data_rules.test_polars_compiler import (
+    _backend,
+    _binding,
+    _published_rule,
+    _schema,
+)
+from tests.runner.test_artifacts import FakeMinio, _store
+
+
+class Resolver:
+    def __init__(self, values):
+        self.values = values
+        self.calls = []
+
+    def resolve(self, *, binding_id, correlation_id):
+        self.calls.append((binding_id, correlation_id))
+        return self.values[binding_id]
+
+
+def _compiled_plan(steps):
+    from app.core.data_rules.compilers.polars import PolarsRuleCompiler
+
+    fields = [
+        ("customer_id", "integer", False),
+        ("name", "string", True),
+        ("mobile", "string", True),
+    ]
+    input_schema = _schema("bd:customer:raw", fields)
+    output_schema = _schema("bd:customer:clean", fields)
+    input_binding = _binding(input_schema, access_mode="read")
+    output_binding = _binding(output_schema, access_mode="write")
+    rule = _published_rule(input_schema, output_schema, steps)
+    compiled = PolarsRuleCompiler().compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=_backend(),
+    )
+    return compiled, input_binding
+
+
+def _node(compiled):
+    return {
+        "id": "task5_polars",
+        "type": "rule.apply",
+        "purpose": "write",
+        "idempotency": {
+            "strategy": "deduplication_key",
+            "key": "customer_id",
+        },
+        "config": {
+            "component_binding_id": new_governance_uid(),
+            "rule_version_id": compiled["plan"]["rule_version_id"],
+            "execution_plan_hash": compiled["plan_hash"],
+        },
+    }
+
+
+def test_polars_adapter_reconstructs_assert_and_deduplicate_and_writes_artifact():
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    compiled, input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            },
+            {
+                "id": "mobile_format",
+                "op": "assert",
+                "expression": "matches(mobile, '^[0-9]{11}$')",
+                "on_failure": "reject",
+                "severity": "error",
+            },
+            {
+                "id": "one_customer",
+                "op": "deduplicate",
+                "keys": ["customer_id"],
+                "order_by": ["name"],
+                "keep": "first",
+            },
+        ]
+    )
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {
+                "customer_id": [1, 1, 2],
+                "name": [" Alice ", "Alice B", " Bad "],
+                "mobile": ["13800138000", "13800138000", "invalid"],
+            }
+        ).lazy(),
+        correlation_id,
+        600,
+    )
+    resolver = Resolver(
+        {
+            input_binding["id"]: {
+                **source,
+                "binding_hash": compiled["plan"]["input_binding_hash"],
+            }
+        }
+    )
+    adapter = PolarsRulePlanAdapter(
+        artifact_store=store,
+        artifact_resolver=resolver,
+        masking_policies={
+            "customer_mobile_last4": "preserve_last_4"
+        },
+        artifact_ttl_seconds=300,
+    )
+
+    result = adapter.execute(
+        plan=compiled["plan"],
+        node=_node(compiled),
+        parameters={},
+        write_authorized=True,
+        correlation_id=correlation_id,
+    )
+
+    assert result["rows_in"] == 3
+    assert result["rows_out"] == 1
+    assert result["rows_rejected"] == 2
+    assert result["violation_count"] == 1
+    assert result["violations"] == [
+        {"step_id": "mobile_format", "count": 1}
+    ]
+    assert result["commit_outcome"] == "committed"
+    assert store.read(
+        result["artifact_ref"], result["digest"]
+    ).collect().to_dicts() == [
+        {
+            "customer_id": 1,
+            "name": "Alice",
+            "mobile": "13800138000",
+        }
+    ]
+
+
+def test_polars_adapter_fails_closed_for_plan_hash_binding_and_authorization():
+    from app.runner.rule_polars import PolarsRulePlanAdapter
+
+    compiled, input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    store = _store(FakeMinio())
+    correlation_id = new_governance_uid()
+    source = store.write(
+        pl.DataFrame(
+            {"customer_id": [1], "name": [" A "], "mobile": ["1"]}
+        ).lazy(),
+        correlation_id,
+        600,
+    )
+    resolver = Resolver(
+        {
+            input_binding["id"]: {
+                **source,
+                "binding_hash": "0" * 64,
+            }
+        }
+    )
+    adapter = PolarsRulePlanAdapter(
+        artifact_store=store,
+        artifact_resolver=resolver,
+    )
+    node = _node(compiled)
+
+    with pytest.raises(NodeExecutionError, match="authorization"):
+        adapter.execute(
+            plan=compiled["plan"],
+            node=node,
+            parameters={},
+            write_authorized=False,
+            correlation_id=correlation_id,
+        )
+    with pytest.raises(NodeExecutionError, match="binding"):
+        adapter.execute(
+            plan=compiled["plan"],
+            node=node,
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+    resolver.values[input_binding["id"]]["binding_hash"] = compiled["plan"][
+        "input_binding_hash"
+    ]
+    node["config"]["execution_plan_hash"] = "0" * 64
+    with pytest.raises(NodeExecutionError, match="hash"):
+        adapter.execute(
+            plan=compiled["plan"],
+            node=node,
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+    tampered = copy.deepcopy(compiled["plan"])
+    tampered["operations"][0]["callable"] = "unsafe"
+    node["config"]["execution_plan_hash"] = compiled["plan_hash"]
+    with pytest.raises(NodeExecutionError, match="invalid"):
+        adapter.execute(
+            plan=tampered,
+            node=node,
+            parameters={},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+
+def test_rule_executor_attests_polars_canonical_hashes_and_forwards_correlation():
+    from app.runner.rules import RulePlanExecutor
+
+    compiled, _input_binding = _compiled_plan(
+        [
+            {
+                "id": "trim_name",
+                "op": "normalize_text",
+                "column": "name",
+                "trim": True,
+            }
+        ]
+    )
+    node = _node(compiled)
+    correlation_id = new_governance_uid()
+    plan = compiled["plan"]
+    record = {
+        "component_binding_id": node["config"]["component_binding_id"],
+        "rule_version_id": plan["rule_version_id"],
+        "backend": "polars_batch",
+        "compiler_version": compiled["compiler_version"],
+        "plan": plan,
+        "plan_hash": compiled["plan_hash"],
+        "schema_hashes": {
+            "rule_spec_hash": plan["rule_spec_hash"],
+            "input_schema_snapshot_id": plan["input_schema_snapshot_id"],
+            "input_schema_hash": plan["input_schema_hash"],
+            "output_schema_snapshot_id": plan["output_schema_snapshot_id"],
+            "output_schema_hash": plan["output_schema_hash"],
+        },
+        "canonical_rule_spec_hash": plan["rule_spec_hash"],
+        "canonical_input_schema_snapshot_id": plan[
+            "input_schema_snapshot_id"
+        ],
+        "canonical_input_schema_hash": plan["input_schema_hash"],
+        "canonical_output_schema_snapshot_id": plan[
+            "output_schema_snapshot_id"
+        ],
+        "canonical_output_schema_hash": plan["output_schema_hash"],
+        "plan_status": "published",
+        "rule_status": "published",
+        "component_kind": "rule.apply",
+        "binding_idempotency": node["idempotency"],
+    }
+
+    class Repository:
+        def load(self, **_kwargs):
+            return record
+
+    class Adapter:
+        def __init__(self):
+            self.kwargs = None
+
+        def execute(self, **kwargs):
+            self.kwargs = kwargs
+            return {"rows_in": 1, "rows_out": 1, "rows_rejected": 0}
+
+    adapter = Adapter()
+    result = RulePlanExecutor(
+        Repository(), adapters={"polars_batch": adapter}
+    ).execute(
+        node,
+        {},
+        write_authorized=True,
+        correlation_id=correlation_id,
+    )
+
+    assert result["rows_out"] == 1
+    assert adapter.kwargs["correlation_id"] == correlation_id
+
+    record["canonical_input_schema_hash"] = "0" * 64
+    with pytest.raises(NodeExecutionError, match="attestation"):
+        RulePlanExecutor(
+            Repository(), adapters={"polars_batch": adapter}
+        ).execute(
+            node,
+            {},
+            write_authorized=True,
+            correlation_id=correlation_id,
+        )
+
+
+def test_node_registry_forwards_trusted_correlation_context():
+    from app.runner.nodes import NodeRegistry
+
+    class Executor:
+        def __init__(self):
+            self.correlation_id = None
+
+        def execute(self, _node, _parameters, **kwargs):
+            self.correlation_id = kwargs["correlation_id"]
+            return {"ok": True}
+
+    executor = Executor()
+    correlation_id = new_governance_uid()
+    assert NodeRegistry({"rule.apply": executor}).execute(
+        {"type": "rule.apply"},
+        {},
+        write_authorized=True,
+        correlation_id=correlation_id,
+    ) == {"ok": True}
+    assert executor.correlation_id == correlation_id