feat: execute governed SQL rule plans

app/core/data_rules/compilers/__init__.py

@@ -0,0 +1,57 @@
+"""Deployment-time compiler selection for governed data rules."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from app.core.data_rules.compilers.base import RuleCompiler
+from app.core.data_rules.contracts import read_rule_spec
+from app.core.data_rules.expressions import backend_support
+
+
+def _dialect(value: Any) -> str:
+    normalized = str(value or "").strip().lower()
+    return "postgresql" if normalized == "postgres" else normalized
+
+
+class CompilerRegistry:
+    """Select only a concrete same-source compiler registered by the runtime."""
+
+    def __init__(self, compilers: Mapping[str, RuleCompiler]):
+        self.compilers = {
+            _dialect(name): compiler for name, compiler in dict(compilers).items()
+        }
+
+    def select(
+        self,
+        rule_spec: dict[str, Any],
+        input_binding: dict[str, Any],
+        output_binding: dict[str, Any],
+    ) -> RuleCompiler:
+        spec = read_rule_spec(rule_spec)
+        if not isinstance(input_binding, dict) or not isinstance(
+            output_binding, dict
+        ):
+            raise ValueError("dataset bindings must be objects")
+        if input_binding.get("data_source_uid") != output_binding.get(
+            "data_source_uid"
+        ):
+            raise ValueError("cross-source rules require a batch compiler")
+        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)
+        if compiler is None or not callable(getattr(compiler, "compile", None)):
+            raise ValueError("no registered compiler matches the dataset bindings")
+        return compiler
+
+
+__all__ = ["CompilerRegistry", "RuleCompiler"]

app/core/data_rules/compilers/base.py

@@ -0,0 +1,23 @@
+"""Closed interfaces for deployment-bound rule compilers."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class RuleCompiler(ABC):
+    """Compile one governed RuleVersion against physical dataset bindings."""
+
+    @abstractmethod
+    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]:
+        raise NotImplementedError

app/core/data_rules/compilers/sql.py

@@ -0,0 +1,829 @@
+"""SQLGlot compiler for one credential-free, deployment-bound SQL rule plan."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import re
+from decimal import Decimal
+from typing import Any
+
+import sqlglot
+from sqlglot import exp
+
+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 (
+    validate_dataset_binding,
+    validate_schema_snapshot,
+)
+from app.core.data_rules.expressions import (
+    backend_support,
+    type_check_expression,
+    validate_rule_expressions,
+)
+
+COMPILER_VERSION = "dataops-sqlglot-30.13.0"
+PLAN_SCHEMA_VERSION = "1.0"
+SUPPORTED_DIALECTS = {"postgresql", "mysql"}
+SUPPORTED_OPERATIONS = {
+    "assert",
+    "cast",
+    "deduplicate",
+    "derive",
+    "fill_null",
+    "filter",
+    "map_values",
+    "normalize_text",
+    "regex_replace",
+}
+CAPABILITY_KEYS = {
+    "dialect",
+    "timezone",
+    "collation",
+    "rounding_mode",
+    "regex_engine",
+}
+PLAN_KEYS = {
+    "schema_version",
+    "dialect",
+    "capabilities",
+    "data_source_uid",
+    "rule_version_id",
+    "input_binding_id",
+    "output_binding_id",
+    "statements",
+    "result_contract",
+}
+RESULT_CONTRACT = {
+    "rows_in": "counted",
+    "rows_out": "counted",
+    "rows_rejected": "counted",
+}
+_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$")
+_PORTABLE_REGEX_FORBIDDEN = re.compile(r"\(\?|\\[1-9]|\\[pP]")
+_NUMERIC_TYPES = {"integer", "decimal", "float", "double"}
+_TYPE_NAMES = {
+    "binary": "BINARY",
+    "boolean": "BOOLEAN",
+    "date": "DATE",
+    "decimal": "DECIMAL",
+    "double": "DOUBLE",
+    "float": "FLOAT",
+    "integer": "BIGINT",
+    "json": "JSON",
+    "string": "TEXT",
+    "timestamp": "TIMESTAMP",
+    "timestamptz": "TIMESTAMPTZ",
+}
+
+
+def _types_compatible(source_type: str, target_type: str) -> bool:
+    return source_type == target_type or {
+        source_type,
+        target_type,
+    } <= _NUMERIC_TYPES
+
+
+def _value_matches_type(value: Any, field_type: str) -> bool:
+    if value is None:
+        return False
+    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
+        if isinstance(value, (int, float)):
+            return True
+        if isinstance(value, str):
+            try:
+                Decimal(value)
+                return True
+            except Exception:
+                return False
+    if field_type in {"date", "timestamp", "timestamptz"}:
+        return isinstance(value, str)
+    return False
+
+
+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 SQL plan must be JSON serializable") from exc
+
+
+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 _dialect(value: Any) -> str:
+    normalized = str(value or "").strip().lower()
+    return "postgresql" if normalized == "postgres" else normalized
+
+
+def _identifier(value: Any, label: str) -> str:
+    normalized = str(value or "")
+    if _IDENTIFIER.fullmatch(normalized) is None:
+        raise ValueError(f"{label} must be a SQL identifier")
+    return normalized
+
+
+def _quoted_identifier(value: str) -> exp.Identifier:
+    return exp.Identifier(this=value, quoted=True)
+
+
+def _column(value: str) -> exp.Column:
+    return exp.Column(this=_quoted_identifier(value))
+
+
+def _table(object_ref: Any) -> exp.Table:
+    parts = str(object_ref or "").split(".")
+    if len(parts) != 2:
+        raise ValueError("table object_ref must be schema.name")
+    schema_name = _identifier(parts[0], "table schema")
+    table_name = _identifier(parts[1], "table name")
+    return exp.Table(
+        this=_quoted_identifier(table_name),
+        db=_quoted_identifier(schema_name),
+    )
+
+
+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")
+    normalized = validate_dataset_binding(
+        {key: item for key, item in value.items() if key != "id"}
+    )
+    normalized["dialect"] = _dialect(normalized["dialect"])
+    return {"id": binding_id, **normalized}
+
+
+def _capabilities(
+    value: Any, *, dialect: str, timezone: str
+) -> dict[str, str]:
+    if not isinstance(value, dict) or set(value) != CAPABILITY_KEYS:
+        raise ValueError("SQL backend capabilities must have a closed shape")
+    capabilities = {
+        key: str(item).strip() for key, item in copy.deepcopy(value).items()
+    }
+    capabilities["dialect"] = _dialect(capabilities["dialect"])
+    if capabilities["dialect"] != dialect:
+        raise ValueError("SQL backend capability dialect does not match binding")
+    if capabilities["timezone"] != timezone:
+        raise ValueError("SQL backend timezone does not preserve rule semantics")
+    if not capabilities["collation"]:
+        raise ValueError("SQL backend collation capability is required")
+    if capabilities["rounding_mode"] != "half_away_from_zero":
+        raise ValueError("SQL backend rounding mode is unsupported")
+    expected_regex = "posix" if dialect == "postgresql" else "icu"
+    if capabilities["regex_engine"] != expected_regex:
+        raise ValueError("SQL backend regex engine is unsupported")
+    return capabilities
+
+
+class _ParameterStore:
+    def __init__(self):
+        self.values: dict[str, Any] = {}
+
+    def add(self, value: Any, *, value_type: str | None = None) -> exp.Expression:
+        name = f"rule_p_{len(self.values) + 1}"
+        if isinstance(value, Decimal):
+            value = str(value)
+        if not isinstance(value, (type(None), bool, int, float, str)):
+            raise ValueError("SQL parameter values must be JSON scalars")
+        self.values[name] = value
+        placeholder: exp.Expression = exp.Placeholder(this=name)
+        if value_type == "decimal":
+            placeholder = exp.Cast(
+                this=placeholder,
+                to=exp.DataType.build("DECIMAL"),
+            )
+        return placeholder
+
+
+class _ExpressionCompiler:
+    def __init__(
+        self,
+        *,
+        dialect: str,
+        fields: dict[str, str],
+        parameters: _ParameterStore,
+    ):
+        self.dialect = dialect
+        self.fields = fields
+        self.parameters = parameters
+
+    def compile(self, ast: dict[str, Any]) -> exp.Expression:
+        type_check_expression(ast, self.fields)
+        if self.dialect not in backend_support(ast):
+            raise ValueError("expression is unsupported by the bound SQL dialect")
+        return self._compile(ast)
+
+    def _compile(self, node: dict[str, Any]) -> exp.Expression:
+        kind = node["kind"]
+        if kind == "identifier":
+            return _column(_identifier(node["name"], "expression field"))
+        if kind == "literal":
+            literal_type = node["type"]
+            if literal_type == "null":
+                return exp.Null()
+            return self.parameters.add(
+                node["value"],
+                value_type=literal_type,
+            )
+        if kind == "unary":
+            operand = self._compile(node["operand"])
+            return (
+                exp.Not(this=operand)
+                if node["operator"] == "!"
+                else exp.Neg(this=operand)
+            )
+        if kind == "binary":
+            left_node = node["left"]
+            right_node = node["right"]
+            operator = node["operator"]
+            left = self._compile(left_node)
+            right = self._compile(right_node)
+            left_null = left_node.get("kind") == "literal" and left_node.get(
+                "type"
+            ) == "null"
+            right_null = right_node.get("kind") == "literal" and right_node.get(
+                "type"
+            ) == "null"
+            if operator in {"==", "!="} and (left_null or right_null):
+                concrete = right if left_null else left
+                predicate = exp.Is(this=concrete, expression=exp.Null())
+                return exp.Not(this=predicate) if operator == "!=" else predicate
+            operators = {
+                "||": exp.Or,
+                "&&": exp.And,
+                "==": exp.EQ,
+                "!=": exp.NEQ,
+                "<": exp.LT,
+                "<=": exp.LTE,
+                ">": exp.GT,
+                ">=": exp.GTE,
+                "+": exp.Add,
+                "-": exp.Sub,
+                "*": exp.Mul,
+                "/": exp.Div,
+                "%": exp.Mod,
+            }
+            return operators[operator](this=left, expression=right)
+        function = node["function"]
+        arguments = [self._compile(item) for item in node["arguments"]]
+        if function == "matches":
+            return exp.RegexpLike(this=arguments[0], expression=arguments[1])
+        if function == "trim":
+            return exp.Trim(this=arguments[0])
+        if function == "length":
+            return exp.Length(this=arguments[0])
+        if function == "coalesce":
+            return exp.Coalesce(this=arguments[0], expressions=arguments[1:])
+        if function == "abs":
+            return exp.Abs(this=arguments[0])
+        if function == "round":
+            return exp.Round(
+                this=arguments[0],
+                decimals=arguments[1] if len(arguments) == 2 else None,
+            )
+        if function == "date":
+            return exp.Cast(
+                this=arguments[0],
+                to=exp.DataType.build("DATE"),
+            )
+        if function == "timestamp" and self.dialect == "postgresql":
+            return exp.Cast(
+                this=arguments[0],
+                to=exp.DataType.build("TIMESTAMPTZ"),
+            )
+        raise ValueError(f"expression function {function} is not SQL executable")
+
+
+def _source_relation(
+    query: exp.Select, alias_number: int
+) -> exp.Subquery:
+    return exp.Subquery(
+        this=query,
+        alias=exp.TableAlias(
+            this=exp.Identifier(this=f"_dataops_s{alias_number}")
+        ),
+    )
+
+
+def _project(
+    source: exp.Expression,
+    fields: list[str],
+    *,
+    replacements: dict[str, exp.Expression] | None = None,
+    where: exp.Expression | None = None,
+) -> exp.Select:
+    replacements = replacements or {}
+    projections = [
+        (replacements.get(name) or _column(name)).as_(
+            _quoted_identifier(name)
+        )
+        for name in fields
+    ]
+    query = exp.Select(expressions=projections).from_(source)
+    if where is not None:
+        query = query.where(where)
+    return query
+
+
+def _portable_regex(pattern: Any) -> str:
+    if not isinstance(pattern, str) or len(pattern) > 500:
+        raise ValueError("regex pattern is invalid")
+    if _PORTABLE_REGEX_FORBIDDEN.search(pattern):
+        raise ValueError("regex pattern is not portable")
+    try:
+        re.compile(pattern)
+    except re.error as exc:
+        raise ValueError("regex pattern is invalid") from exc
+    return pattern
+
+
+def _render(expression: exp.Expression, dialect: str) -> str:
+    sql_dialect = "postgres" if dialect == "postgresql" else dialect
+    statement = expression.sql(dialect=sql_dialect)
+    # SQLGlot renders PostgreSQL placeholders using DBAPI pyformat. The
+    # Runner stores SQLAlchemy named binds, so normalize only compiler-owned
+    # placeholder tokens after AST generation.
+    if dialect == "postgresql":
+        statement = re.sub(
+            r"%\(([A-Za-z_][A-Za-z0-9_]*)\)s",
+            r":\1",
+            statement,
+        )
+    return statement
+
+
+def validate_bound_sql_plan(value: Any) -> dict[str, Any]:
+    """Validate the exact executable plan shape and its SQLGlot AST."""
+
+    if not isinstance(value, dict) or set(value) != PLAN_KEYS:
+        raise ValueError("bound SQL plan must have a closed shape")
+    plan = copy.deepcopy(value)
+    if plan["schema_version"] != PLAN_SCHEMA_VERSION:
+        raise ValueError("unsupported bound SQL plan schema version")
+    dialect = _dialect(plan["dialect"])
+    if dialect not in SUPPORTED_DIALECTS:
+        raise ValueError("unsupported bound SQL plan dialect")
+    plan["dialect"] = dialect
+    plan["capabilities"] = _capabilities(
+        plan["capabilities"],
+        dialect=dialect,
+        timezone=str(plan["capabilities"].get("timezone") or ""),
+    )
+    for key in (
+        "data_source_uid",
+        "rule_version_id",
+        "input_binding_id",
+        "output_binding_id",
+    ):
+        plan[key] = _uid(plan[key], key)
+    statements = plan["statements"]
+    if not isinstance(statements, list) or len(statements) != 1:
+        raise ValueError("bound SQL plan must contain one statement")
+    statement = statements[0]
+    if not isinstance(statement, dict) or set(statement) != {
+        "purpose",
+        "sql",
+        "parameters",
+    }:
+        raise ValueError("bound SQL statement must have a closed shape")
+    if statement["purpose"] != "write":
+        raise ValueError("bound SQL plan must contain a write statement")
+    sql = statement["sql"]
+    if (
+        not isinstance(sql, str)
+        or not sql.strip()
+        or ";" in sql
+        or len(sql) > 100_000
+    ):
+        raise ValueError("bound SQL statement must be one bounded statement")
+    sql_dialect = "postgres" if dialect == "postgresql" else dialect
+    try:
+        expressions = sqlglot.parse(sql, read=sql_dialect)
+    except sqlglot.errors.ParseError as exc:
+        raise ValueError("bound SQL statement is invalid") from exc
+    if len(expressions) != 1 or not isinstance(expressions[0], exp.Insert):
+        raise ValueError("bound SQL statement must be one INSERT")
+    parsed = expressions[0]
+    if any(
+        isinstance(node, (exp.Command, exp.Delete, exp.Update))
+        for node in parsed.walk()
+    ):
+        raise ValueError("bound SQL statement contains unsupported operations")
+    source_tables = list(parsed.expression.find_all(exp.Table))
+    if len(source_tables) != 1:
+        raise ValueError("bound SQL plan must read exactly one source table")
+    parameters = statement["parameters"]
+    if not isinstance(parameters, dict) or len(parameters) > 500:
+        raise ValueError("bound SQL parameters must be a bounded object")
+    for name, item in parameters.items():
+        if _IDENTIFIER.fullmatch(str(name)) is None or not isinstance(
+            item, (type(None), bool, int, float, str)
+        ):
+            raise ValueError("bound SQL parameters are invalid")
+    placeholders = {
+        placeholder.name for placeholder in parsed.find_all(exp.Placeholder)
+    }
+    if placeholders != set(parameters):
+        raise ValueError("bound SQL placeholders do not match parameters")
+    if plan["result_contract"] != RESULT_CONTRACT:
+        raise ValueError("bound SQL result contract is unsupported")
+    _canonical_json(plan)
+    return plan
+
+
+def bound_sql_plan_hash(value: Any) -> str:
+    plan = validate_bound_sql_plan(value)
+    return hashlib.sha256(_canonical_json(plan).encode("utf-8")).hexdigest()
+
+
+def bound_sql_plan_relations(value: Any) -> dict[str, str]:
+    """Return the two physical relation names from a validated INSERT plan."""
+
+    plan = validate_bound_sql_plan(value)
+    sql_dialect = (
+        "postgres" if plan["dialect"] == "postgresql" else plan["dialect"]
+    )
+    insert = sqlglot.parse_one(
+        plan["statements"][0]["sql"],
+        read=sql_dialect,
+    )
+    target = insert.this
+    if not isinstance(target, exp.Schema) or not isinstance(
+        target.this, exp.Table
+    ):
+        raise ValueError("bound SQL target relation is invalid")
+    sources = list(insert.expression.find_all(exp.Table))
+    if len(sources) != 1:
+        raise ValueError("bound SQL source relation is invalid")
+
+    def object_ref(table: exp.Table) -> str:
+        if not table.db or not table.name:
+            raise ValueError("bound SQL relation must be schema-qualified")
+        return f"{table.db}.{table.name}"
+
+    return {
+        "input_object_ref": object_ref(sources[0]),
+        "output_object_ref": object_ref(target.this),
+    }
+
+
+class SqlGlotRuleCompiler(RuleCompiler):
+    """Compile the proven first operator slice into one INSERT ... SELECT."""
+
+    def __init__(self, dialect: str):
+        self.dialect = _dialect(dialect)
+        if self.dialect not in SUPPORTED_DIALECTS:
+            raise ValueError("unsupported SQLGlot compiler dialect")
+
+    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")
+        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["data_source_uid"] != target_binding["data_source_uid"]:
+            raise ValueError("cross-source rules require a batch compiler")
+        if source_binding["dialect"] != self.dialect or target_binding[
+            "dialect"
+        ] != self.dialect:
+            raise ValueError("binding dialect does not match compiler")
+        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"] not in {"table", "view"} or target_binding[
+            "object_kind"
+        ] != "table":
+            raise ValueError("SQL rules require a table/view input and table output")
+        if target_binding["write_mode"] != "append":
+            raise ValueError("unsupported SQL output write mode")
+        capabilities = _capabilities(
+            backend,
+            dialect=self.dialect,
+            timezone=spec["timezone"],
+        )
+
+        field_types = {
+            field["name"]: field["type"] for field in source_schema["fields"]
+        }
+        supported = validate_rule_expressions(spec, field_types)
+        if self.dialect not in supported:
+            raise ValueError("rule expressions are unsupported by the bound dialect")
+        input_fields = [field["name"] for field in source_schema["fields"]]
+        output_fields = [field["name"] for field in target_schema["fields"]]
+        current_fields = list(input_fields)
+        source: exp.Expression = _table(source_binding["object_ref"])
+        parameters = _ParameterStore()
+        expression_compiler = _ExpressionCompiler(
+            dialect=self.dialect,
+            fields=field_types,
+            parameters=parameters,
+        )
+        alias_number = 0
+
+        def replace(
+            replacements: dict[str, exp.Expression],
+            *,
+            where: exp.Expression | None = None,
+        ) -> None:
+            nonlocal source, alias_number
+            query = _project(
+                source,
+                current_fields,
+                replacements=replacements,
+                where=where,
+            )
+            alias_number += 1
+            source = _source_relation(query, alias_number)
+
+        for step in spec["steps"]:
+            operation = step["op"]
+            if operation not in SUPPORTED_OPERATIONS:
+                raise ValueError(
+                    f"unsupported SQL rule operation: {operation}"
+                )
+            if operation in {
+                "cast",
+                "fill_null",
+                "map_values",
+                "normalize_text",
+                "regex_replace",
+            }:
+                column_name = _identifier(step.get("column"), f"{operation} column")
+                if column_name not in current_fields:
+                    raise ValueError(f"unknown SQL rule column: {column_name}")
+            if operation == "cast":
+                target_type = str(step.get("to") or "").strip().lower()
+                type_name = _TYPE_NAMES.get(target_type)
+                if type_name is None:
+                    raise ValueError("unsupported SQL cast target")
+                replace(
+                    {
+                        column_name: exp.Cast(
+                            this=_column(column_name),
+                            to=exp.DataType.build(type_name),
+                        )
+                    }
+                )
+                field_types[column_name] = target_type
+            elif operation == "normalize_text":
+                if field_types[column_name] != "string":
+                    raise ValueError("normalize_text requires a string field type")
+                if step.get("lowercase") or step.get("uppercase"):
+                    raise ValueError(
+                        "SQL case normalization requires a proven Unicode collation"
+                    )
+                if not step.get("trim"):
+                    raise ValueError("normalize_text has no proven SQL operation")
+                replace({column_name: exp.Trim(this=_column(column_name))})
+            elif operation == "regex_replace":
+                if field_types[column_name] != "string":
+                    raise ValueError("regex_replace requires a string field type")
+                pattern = _portable_regex(step.get("pattern"))
+                replacement = step.get("replacement")
+                if not isinstance(replacement, str):
+                    raise ValueError("regex replacement must be a string")
+                replace(
+                    {
+                        column_name: exp.RegexpReplace(
+                            this=_column(column_name),
+                            expression=parameters.add(pattern),
+                            replacement=parameters.add(replacement),
+                        )
+                    }
+                )
+            elif operation == "fill_null":
+                if not _value_matches_type(
+                    step.get("value"),
+                    field_types[column_name],
+                ):
+                    raise ValueError(
+                        "fill_null value does not match the field type"
+                    )
+                replace(
+                    {
+                        column_name: exp.Coalesce(
+                            this=_column(column_name),
+                            expressions=[parameters.add(step.get("value"))],
+                        )
+                    }
+                )
+            elif operation in {"filter", "assert"}:
+                if operation == "assert" and step.get("on_failure") != "reject":
+                    raise ValueError("assert failure action is not SQL executable")
+                predicate = expression_compiler.compile(step["expression_ast"])
+                replace({}, where=predicate)
+            elif operation == "derive":
+                target = _identifier(step.get("target"), "derive target")
+                if target not in current_fields and target not in output_fields:
+                    raise ValueError("derive target is not in the output schema")
+                result_type = type_check_expression(
+                    step["expression_ast"],
+                    field_types,
+                )
+                target_type = next(
+                    field["type"]
+                    for field in target_schema["fields"]
+                    if field["name"] == target
+                )
+                if not _types_compatible(result_type, target_type):
+                    raise ValueError(
+                        "derive expression type does not match its target type"
+                    )
+                derived = expression_compiler.compile(step["expression_ast"])
+                if target not in current_fields:
+                    current_fields.append(target)
+                replace({target: derived})
+                field_types[target] = target_type
+            elif operation == "map_values":
+                if field_types[column_name] != "string":
+                    raise ValueError("map_values requires a string field type")
+                mapping = step.get("mapping")
+                if not isinstance(mapping, dict) or not mapping:
+                    raise ValueError("map_values mapping must be non-empty")
+                if not all(
+                    isinstance(source_value, str)
+                    and isinstance(target_value, str)
+                    for source_value, target_value in mapping.items()
+                ):
+                    raise ValueError(
+                        "map_values values must match the string field type"
+                    )
+                case = exp.Case()
+                for source_value, target_value in sorted(
+                    mapping.items(), key=lambda item: str(item[0])
+                ):
+                    case = case.when(
+                        exp.EQ(
+                            this=_column(column_name),
+                            expression=parameters.add(source_value),
+                        ),
+                        parameters.add(target_value),
+                    )
+                case = case.else_(_column(column_name))
+                replace({column_name: case})
+            elif operation == "deduplicate":
+                keys = [
+                    _identifier(item, "deduplicate key")
+                    for item in step.get("keys", [])
+                ]
+                order_by = [
+                    _identifier(item, "deduplicate order field")
+                    for item in step.get("order_by", [])
+                ]
+                if (
+                    not keys
+                    or not order_by
+                    or not set(keys + order_by) <= set(current_fields)
+                ):
+                    raise ValueError(
+                        "deduplicate requires known keys and deterministic order"
+                    )
+                descending = step.get("keep", "first") == "last"
+                row_number = exp.Window(
+                    this=exp.RowNumber(),
+                    partition_by=[_column(item) for item in keys],
+                    order=exp.Order(
+                        expressions=[
+                            exp.Ordered(
+                                this=_column(item),
+                                desc=descending,
+                            )
+                            for item in order_by
+                        ]
+                    ),
+                ).as_("_dataops_row_number")
+                ranked = exp.Select(
+                    expressions=[
+                        *[
+                            _column(name).as_(_quoted_identifier(name))
+                            for name in current_fields
+                        ],
+                        row_number,
+                    ]
+                ).from_(source)
+                alias_number += 1
+                ranked_source = _source_relation(ranked, alias_number)
+                deduplicated = _project(
+                    ranked_source,
+                    current_fields,
+                    where=exp.EQ(
+                        this=_column("_dataops_row_number"),
+                        expression=exp.Literal.number(1),
+                    ),
+                )
+                alias_number += 1
+                source = _source_relation(deduplicated, alias_number)
+
+        if not set(output_fields) <= set(current_fields):
+            raise ValueError("compiled SQL cannot produce the output schema")
+        output_types = {
+            field["name"]: field["type"] for field in target_schema["fields"]
+        }
+        if any(
+            not _types_compatible(field_types[name], output_types[name])
+            for name in output_fields
+        ):
+            raise ValueError(
+                "compiled SQL field types do not match the output schema"
+            )
+        final_select = exp.Select(
+            expressions=[_column(name) for name in output_fields]
+        ).from_(source)
+        target = exp.Schema(
+            this=_table(target_binding["object_ref"]),
+            expressions=[_quoted_identifier(name) for name in output_fields],
+        )
+        statement_ast = exp.Insert(this=target, expression=final_select)
+        statement = _render(statement_ast, self.dialect)
+        plan = validate_bound_sql_plan(
+            {
+                "schema_version": PLAN_SCHEMA_VERSION,
+                "dialect": self.dialect,
+                "capabilities": capabilities,
+                "data_source_uid": source_binding["data_source_uid"],
+                "rule_version_id": rule_version_id,
+                "input_binding_id": source_binding["id"],
+                "output_binding_id": target_binding["id"],
+                "statements": [
+                    {
+                        "purpose": "write",
+                        "sql": statement,
+                        "parameters": parameters.values,
+                    }
+                ],
+                "result_contract": RESULT_CONTRACT,
+            }
+        )
+        return {
+            "backend": "sql_pushdown",
+            "compiler_version": COMPILER_VERSION,
+            "status": "compiled",
+            "plan": plan,
+            "plan_hash": bound_sql_plan_hash(plan),
+        }
+
+
+__all__ = [
+    "COMPILER_VERSION",
+    "SqlGlotRuleCompiler",
+    "bound_sql_plan_hash",
+    "bound_sql_plan_relations",
+    "validate_bound_sql_plan",
+]

app/core/data_rules/release.py

@@ -214,3 +214,58 @@ class ProductionLineReleaseService:
             dataflow_version_id=version_id,
             package=package,
         )
+
+
+class BoundSqlPlanService:
+    """Compile and persist a physical plan only after deployment bindings exist.
+
+    Compilation is deliberately persisted as ``compiled``. Test evidence and
+    publication transitions remain separate lifecycle actions (Task 7), so
+    this vertical slice never labels an unexecuted plan tested or published.
+    """
+
+    def __init__(self, repository, compiler_registry):
+        self.repository = repository
+        self.compiler_registry = compiler_registry
+
+    def compile_and_persist(
+        self,
+        *,
+        component_binding_id: str,
+        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]:
+        component_id = _uid(
+            component_binding_id, "component_binding_id"
+        )
+        compiler = self.compiler_registry.select(
+            rule_version.get("rule_spec"),
+            input_binding,
+            output_binding,
+        )
+        compiled = compiler.compile(
+            rule_version=rule_version,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=input_binding,
+            output_binding=output_binding,
+            backend=backend,
+        )
+        return self.repository.persist_bound_component_plan(
+            component_binding_id=component_id,
+            rule_version_id=_uid(
+                rule_version.get("id"), "rule_version_id"
+            ),
+            input_binding_id=_uid(
+                input_binding.get("id"), "input_binding_id"
+            ),
+            output_binding_id=_uid(
+                output_binding.get("id"), "output_binding_id"
+            ),
+            compiled=compiled,
+            status="compiled",
+        )

app/core/data_rules/repository.py

@@ -2,8 +2,8 @@
 
 from __future__ import annotations
 
-import json
 import hashlib
+import json
 import re
 from typing import Any
 
@@ -28,7 +28,6 @@ from app.core.data_rules.execution_contracts import (
     validate_schema_snapshot,
 )
 
-
 RULE_CATEGORIES = {
     "general",
     "reusable",
@@ -879,6 +878,156 @@ class DataRuleRepository:
             },
         )
 
+    def persist_bound_component_plan(
+        self,
+        *,
+        component_binding_id: str,
+        rule_version_id: str,
+        input_binding_id: str,
+        output_binding_id: str,
+        compiled: dict[str, Any],
+        status: str = "compiled",
+    ) -> dict[str, Any]:
+        """Persist a deployment-bound SQL 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"
+            )
+        if not isinstance(compiled, dict) or set(compiled) != {
+            "backend",
+            "compiler_version",
+            "status",
+            "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")
+        compiler_version = _text(
+            compiled["compiler_version"], "compiler_version", 80
+        )
+        plan = validate_bound_sql_plan(compiled["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 (
+            plan["rule_version_id"] != rule_id
+            or plan["input_binding_id"] != input_id
+            or plan["output_binding_id"] != output_id
+        ):
+            raise ValueError("compiled bound SQL plan identifiers do not match")
+
+        linkage = (
+            self.session.execute(
+                text(
+                    "SELECT cb.id::text AS component_binding_id, "
+                    "ins.schema_hash AS input_schema_hash, "
+                    "outs.schema_hash AS output_schema_hash, "
+                    "ib.object_ref AS input_object_ref, "
+                    "ob.object_ref AS output_object_ref, "
+                    "ib.data_source_uid::text AS data_source_uid, "
+                    "ib.dialect AS input_dialect, "
+                    "ob.dialect AS output_dialect "
+                    "FROM public.dataflow_component_bindings cb "
+                    "JOIN public.dataflow_deployments d "
+                    "ON d.dataflow_version_id = cb.dataflow_version_id "
+                    "JOIN public.dataflow_dataset_bindings ib "
+                    "ON ib.dataflow_deployment_id = d.id "
+                    "JOIN public.data_schema_snapshots ins "
+                    "ON ins.id = ib.schema_snapshot_id "
+                    "JOIN public.dataflow_dataset_bindings ob "
+                    "ON ob.dataflow_deployment_id = d.id "
+                    "JOIN public.data_schema_snapshots outs "
+                    "ON outs.id = ob.schema_snapshot_id "
+                    "WHERE cb.id = CAST(:component_binding_id AS uuid) "
+                    "AND cb.rule_version_id = CAST(:rule_version_id AS uuid) "
+                    "AND ib.id = CAST(:input_binding_id AS uuid) "
+                    "AND ob.id = CAST(:output_binding_id AS uuid) "
+                    "/* bound_plan_linkage */"
+                ),
+                {
+                    "component_binding_id": component_id,
+                    "rule_version_id": rule_id,
+                    "input_binding_id": input_id,
+                    "output_binding_id": output_id,
+                },
+            )
+            .mappings()
+            .one_or_none()
+        )
+        if linkage is None:
+            raise ValueError(
+                "bound SQL 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"]
+        ):
+            raise ValueError(
+                "bound SQL plan does not match its physical dataset bindings"
+            )
+        row = (
+            self.session.execute(
+                text(
+                    "INSERT INTO public.rule_execution_plans "
+                    "(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), "
+                    ":plan_hash, CAST(:schema_hashes AS jsonb), :status) "
+                    "RETURNING id::text AS id, status"
+                ),
+                {
+                    "id": new_governance_uid(),
+                    "component_binding_id": component_id,
+                    "compiler_version": compiler_version,
+                    "plan": _json(plan),
+                    "plan_hash": plan_hash,
+                    "schema_hashes": _json(
+                        {
+                            "input": _digest(
+                                linkage["input_schema_hash"],
+                                "input_schema_hash",
+                            ),
+                            "output": _digest(
+                                linkage["output_schema_hash"],
+                                "output_schema_hash",
+                            ),
+                        }
+                    ),
+                    "status": status,
+                },
+            )
+            .mappings()
+            .one()
+        )
+        return {
+            "id": str(row["id"]),
+            "status": str(row["status"]),
+            "backend": "sql_pushdown",
+            "plan_hash": plan_hash,
+        }
+
     def complete_dataflow_release(
         self,
         *,

app/runner/bootstrap.py

@@ -19,10 +19,13 @@ from app.runner.nodes import (
     SqlExecuteExecutor,
     SqlQueryExecutor,
 )
+from app.runner.rule_sql import (
+    SqlGlotQualityPlanAdapter,
+    SqlGlotRulePlanAdapter,
+)
 from app.runner.rules import (
     PostgresRulePlanRepository,
     RulePlanExecutor,
-    SqlRulePlanAdapter,
 )
 
 
@@ -120,15 +123,12 @@ def build_runner_application(settings=None):
         max_rows=settings.max_query_rows,
     )
     write_executor = SqlExecuteExecutor(runtime.manager)
-    sql_rule_adapter = SqlRulePlanAdapter(
-        query_executor=query_executor,
-        write_executor=write_executor,
-    )
+    sql_rule_adapter = SqlGlotRulePlanAdapter(runtime.manager)
     rule_executor = RulePlanExecutor(
         PostgresRulePlanRepository(runtime.platform_engine),
         adapters={
             "sql_pushdown": sql_rule_adapter,
-            "quality_check": sql_rule_adapter,
+            "quality_check": SqlGlotQualityPlanAdapter(),
         },
     )
     registry = NodeRegistry(

app/runner/rule_sql.py

@@ -0,0 +1,259 @@
+"""Runner adapter for exact, published SQLGlot rule plans."""
+
+from __future__ import annotations
+
+import copy
+import re
+
+from sqlalchemy import text
+from sqlglot import exp, parse_one
+
+from app.core.data_rules.compilers.sql import (
+    bound_sql_plan_hash,
+    validate_bound_sql_plan,
+)
+from app.core.data_source.errors import DataSourceWriteOutcomeUnknown
+from app.runner.nodes import NodeExecutionError
+
+
+def _dialect(value):
+    normalized = str(value or "").strip().lower()
+    return "postgresql" if normalized == "postgres" else normalized
+
+
+def _source_count_statement(plan):
+    dialect = "postgres" if plan["dialect"] == "postgresql" else plan["dialect"]
+    insert = parse_one(plan["statements"][0]["sql"], read=dialect)
+    source_tables = list(insert.expression.find_all(exp.Table))
+    if len(source_tables) != 1:
+        raise NodeExecutionError("published SQL rule plan has an invalid source")
+    count = exp.Select(
+        expressions=[exp.Count(this=exp.Star())]
+    ).from_(copy.deepcopy(source_tables[0]))
+    return count.sql(dialect=dialect)
+
+
+def _accepted_count_statement(plan):
+    dialect = "postgres" if plan["dialect"] == "postgresql" else plan["dialect"]
+    insert = parse_one(plan["statements"][0]["sql"], read=dialect)
+    accepted = exp.Select(
+        expressions=[exp.Count(this=exp.Star())]
+    ).from_(
+        exp.Subquery(
+            this=copy.deepcopy(insert.expression),
+            alias=exp.TableAlias(
+                this=exp.Identifier(this="_dataops_accepted")
+            ),
+        )
+    )
+    statement = accepted.sql(dialect=dialect)
+    if plan["dialect"] == "postgresql":
+        statement = re.sub(
+            r"%\(([A-Za-z_][A-Za-z0-9_]*)\)s",
+            r":\1",
+            statement,
+        )
+    return statement
+
+
+def _upsert_statement(plan, key):
+    if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,127}", str(key or "")) is None:
+        raise NodeExecutionError("governed SQL rule idempotency key is invalid")
+    dialect = "postgres" if plan["dialect"] == "postgresql" else plan["dialect"]
+    insert = parse_one(plan["statements"][0]["sql"], read=dialect)
+    target = insert.this
+    if not isinstance(target, exp.Schema):
+        raise NodeExecutionError("published SQL rule target schema is invalid")
+    columns = [column.name for column in target.expressions]
+    if key not in columns:
+        raise NodeExecutionError(
+            "governed SQL rule idempotency key is not an output field"
+        )
+    assignments = []
+    if plan["dialect"] == "postgresql":
+        for name in columns:
+            if name == key:
+                continue
+            assignments.append(
+                exp.EQ(
+                    this=exp.Column(
+                        this=exp.Identifier(this=name, quoted=True)
+                    ),
+                    expression=exp.Column(
+                        this=exp.Identifier(this=name, quoted=True),
+                        table=exp.Identifier(this="EXCLUDED"),
+                    ),
+                )
+            )
+        insert.set(
+            "conflict",
+            exp.OnConflict(
+                duplicate=False,
+                expressions=assignments,
+                action=exp.Var(
+                    this="DO UPDATE" if assignments else "DO NOTHING"
+                ),
+                conflict_keys=[
+                    exp.Ordered(
+                        this=exp.Column(
+                            this=exp.Identifier(this=key, quoted=True)
+                        )
+                    )
+                ],
+            ),
+        )
+    else:
+        for name in columns:
+            assignments.append(
+                exp.EQ(
+                    this=exp.Column(
+                        this=exp.Identifier(this=name, quoted=True)
+                    ),
+                    expression=exp.Anonymous(
+                        this="VALUES",
+                        expressions=[
+                            exp.Identifier(this=name, quoted=True)
+                        ],
+                    ),
+                )
+            )
+        insert.set(
+            "conflict",
+            exp.OnConflict(
+                duplicate=True,
+                expressions=assignments,
+                action=exp.Var(this="UPDATE"),
+            ),
+        )
+    statement = insert.sql(dialect=dialect)
+    if plan["dialect"] == "postgresql":
+        statement = re.sub(
+            r"%\(([A-Za-z_][A-Za-z0-9_]*)\)s",
+            r":\1",
+            statement,
+        )
+    return statement
+
+
+class SqlGlotRulePlanAdapter:
+    """Execute a validated INSERT plan and count its source in one transaction."""
+
+    def __init__(self, manager):
+        self.manager = manager
+
+    def execute(
+        self,
+        *,
+        plan,
+        node,
+        parameters,
+        write_authorized,
+    ):
+        try:
+            normalized = validate_bound_sql_plan(plan)
+        except ValueError as exc:
+            raise NodeExecutionError("published SQL rule plan is invalid") from exc
+        config = node.get("config") or {}
+        if config.get("execution_plan_hash") != bound_sql_plan_hash(normalized):
+            raise NodeExecutionError("published SQL rule plan hash does not match")
+        if config.get("rule_version_id") != normalized["rule_version_id"]:
+            raise NodeExecutionError("published SQL rule plan 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") != "upsert"
+            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 SQL rule plans do not accept unbound runtime parameters"
+            )
+        definition = self.manager.definitions.get(
+            normalized["data_source_uid"]
+        )
+        if definition is None:
+            raise NodeExecutionError("published SQL rule datasource was not found")
+        if _dialect(getattr(definition, "database_type", None)) != normalized[
+            "dialect"
+        ]:
+            raise NodeExecutionError(
+                "published SQL rule datasource dialect does not match"
+            )
+        datasource_capabilities = dict(
+            getattr(definition, "extra_properties", {}) or {}
+        ).get("sql_rule_capabilities")
+        if datasource_capabilities != normalized["capabilities"]:
+            raise NodeExecutionError(
+                "published SQL rule datasource capabilities do not match"
+            )
+
+        statement = normalized["statements"][0]
+        try:
+            with self.manager.connect(
+                normalized["data_source_uid"],
+                purpose="dataflow_write",
+            ) as connection:
+                rows_in = int(
+                    connection.execute(
+                        text(_source_count_statement(normalized)),
+                        {},
+                    ).scalar_one()
+                    or 0
+                )
+                rows_out = int(
+                    connection.execute(
+                        text(_accepted_count_statement(normalized)),
+                        statement["parameters"],
+                    ).scalar_one()
+                    or 0
+                )
+                result = connection.execute(
+                    text(
+                        _upsert_statement(
+                            normalized,
+                            idempotency["key"],
+                        )
+                    ),
+                    statement["parameters"],
+                )
+                if result.rowcount is not None and int(result.rowcount) < 0:
+                    raise NodeExecutionError(
+                        "governed SQL rule returned an invalid write count"
+                    )
+                metrics = {
+                    "rows_in": rows_in,
+                    "rows_out": rows_out,
+                    "rows_rejected": max(0, rows_in - rows_out),
+                    "commit_outcome": "committed",
+                }
+        except DataSourceWriteOutcomeUnknown as exc:
+            raise NodeExecutionError(
+                "governed SQL rule commit outcome is unknown",
+                commit_outcome="unknown",
+            ) from exc
+        except NodeExecutionError:
+            raise
+        except Exception as exc:
+            raise NodeExecutionError(
+                "governed SQL rule write failed",
+                commit_outcome="not_committed",
+            ) from exc
+        return metrics
+
+
+class SqlGlotQualityPlanAdapter:
+    """Explicit fail-closed placeholder until a read-only quality plan exists."""
+
+    def execute(self, **_kwargs):
+        raise NodeExecutionError(
+            "quality_check requires an explicit read-only compiled plan"
+        )
+
+
+__all__ = ["SqlGlotQualityPlanAdapter", "SqlGlotRulePlanAdapter"]

app/runner/rules.py

@@ -5,14 +5,14 @@ from __future__ import annotations
 import hashlib
 import json
 import re
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from sqlalchemy import text
 
 from app.core.common.identifiers import ensure_governance_uid
 from app.runner.nodes import NodeExecutionError
 
-
 CONFIG_KEYS = {
     "component_binding_id",
     "rule_version_id",
@@ -70,7 +70,9 @@ class PostgresRulePlanRepository:
                 p.plan,
                 p.plan_hash,
                 p.status AS plan_status,
-                r.status AS rule_status
+                r.status AS rule_status,
+                b.component_kind,
+                b.idempotency AS binding_idempotency
             FROM public.rule_execution_plans p
             JOIN public.dataflow_component_bindings b
               ON b.id = p.component_binding_id
@@ -93,53 +95,6 @@ class PostgresRulePlanRepository:
         return dict(row) if row is not None else None
 
 
-class SqlRulePlanAdapter:
-    """Execute a compiled, parameterized SQL plan using existing governed nodes."""
-
-    def __init__(self, *, query_executor, write_executor):
-        self.query_executor = query_executor
-        self.write_executor = write_executor
-
-    def execute(
-        self,
-        *,
-        plan,
-        node,
-        parameters,
-        write_authorized,
-    ):
-        if not isinstance(plan, dict):
-            raise NodeExecutionError("published SQL rule plan is invalid")
-        unknown = set(plan) - {
-            "statement",
-            "parameters",
-            "data_source_uid",
-        }
-        if unknown:
-            raise NodeExecutionError(
-                "published SQL rule plan contains unsupported fields"
-            )
-        compiled_node = {
-            "id": node.get("id"),
-            "data_source_uid": _uid(
-                plan.get("data_source_uid"), "plan data_source_uid"
-            ),
-            "purpose": node.get("purpose"),
-            "config": {
-                "statement": plan.get("statement"),
-                "parameters": plan.get("parameters", {}),
-            },
-        }
-        if node.get("type") == "rule.apply":
-            compiled_node["idempotency"] = node.get("idempotency")
-            return self.write_executor.execute(
-                compiled_node,
-                parameters,
-                write_authorized=write_authorized,
-            )
-        return self.query_executor.execute(compiled_node, parameters)
-
-
 class RulePlanExecutor:
     """Fail closed unless the exact published plan is still executable."""
 
@@ -205,6 +160,13 @@ class RulePlanExecutor:
             or _canonical_hash(record.get("plan")) != plan_hash
         ):
             raise NodeExecutionError("published rule plan is not executable")
+        if node.get("type") == "rule.apply" and (
+            record.get("component_kind") != "rule.apply"
+            or record.get("binding_idempotency") != node.get("idempotency")
+        ):
+            raise NodeExecutionError(
+                "governed rule idempotency does not match its binding"
+            )
         backend = record.get("backend")
         adapter = self.adapters.get(backend)
         if adapter is None or not callable(getattr(adapter, "execute", None)):

requirements.txt

@@ -10,6 +10,7 @@ SQLAlchemy==2.0.23
 alembic==1.13.0
 psycopg2-binary==2.9.9
 PyMySQL==1.1.1
+sqlglot==30.13.0
 neo4j==5.26.0
 argon2-cffi==25.1.0
 PyJWT==2.10.1

tests/core/data_rules/test_sql_compiler.py

@@ -0,0 +1,503 @@
+from __future__ import annotations
+
+import copy
+import re
+
+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:task4",
+    }
+
+
+def customer_schemas():
+    fields = [
+        ("customer_id", "integer", False),
+        ("name", "string", True),
+        ("mobile", "string", True),
+        ("balance", "decimal", True),
+    ]
+    return _schema("bd:customer:raw", fields), _schema("bd:customer:clean", fields)
+
+
+def table_binding(schema, object_ref, *, dialect="postgresql", access_mode="read"):
+    return {
+        "id": new_governance_uid(),
+        "data_source_uid": DATA_SOURCE_UID,
+        "object_kind": "table",
+        "object_ref": object_ref,
+        "schema_snapshot_id": schema["id"],
+        "access_mode": access_mode,
+        "dialect": dialect,
+        "write_mode": "append",
+    }
+
+
+def backend(dialect="postgresql"):
+    return {
+        "dialect": dialect,
+        "timezone": "Asia/Shanghai",
+        "collation": "C" if dialect == "postgresql" else "utf8mb4_0900_bin",
+        "rounding_mode": "half_away_from_zero",
+        "regex_engine": "posix" if dialect == "postgresql" else "icu",
+    }
+
+
+def published_rule(steps):
+    spec = validate_rule_spec(
+        {
+            "schema_version": "2.0",
+            "rule_uid": new_governance_uid(),
+            "name": "task4_customer_rule",
+            "input_schema_ref": "bd:customer:raw",
+            "output_schema_ref": "bd:customer:clean",
+            "steps": steps,
+            "null_policy": "explicit",
+            "timezone": "Asia/Shanghai",
+        }
+    )
+    return {
+        "id": new_governance_uid(),
+        "status": "published",
+        "rule_spec": spec,
+        "spec_hash": rule_spec_hash(spec),
+    }
+
+
+DATA_SOURCE_UID = new_governance_uid()
+
+
+def _compile(dialect="postgresql", *, steps=None):
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    input_schema, output_schema = customer_schemas()
+    input_binding = table_binding(
+        input_schema,
+        "raw.customer",
+        dialect=dialect,
+    )
+    output_binding = table_binding(
+        output_schema,
+        "clean.customer",
+        dialect=dialect,
+        access_mode="write",
+    )
+    rule = published_rule(
+        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",
+            },
+        ]
+    )
+    return SqlGlotRuleCompiler(dialect).compile(
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=input_binding,
+        output_binding=output_binding,
+        backend=backend(dialect),
+    )
+
+
+@pytest.mark.parametrize(
+    ("dialect", "source", "target"),
+    [
+        ("postgresql", '"raw"."customer"', '"clean"."customer"'),
+        ("mysql", "`raw`.`customer`", "`clean`.`customer`"),
+    ],
+)
+def test_sql_compiler_emits_one_strict_parameterized_statement(
+    dialect, source, target
+):
+    from app.core.data_rules.compilers.sql import bound_sql_plan_hash
+
+    compiled = _compile(dialect)
+
+    assert compiled["backend"] == "sql_pushdown"
+    assert compiled["plan"]["dialect"] == dialect
+    assert len(compiled["plan"]["statements"]) == 1
+    statement = compiled["plan"]["statements"][0]
+    assert statement["purpose"] == "write"
+    assert ";" not in statement["sql"]
+    assert source in statement["sql"]
+    assert target in statement["sql"]
+    assert "^[0-9]{11}$" not in statement["sql"]
+    assert statement["parameters"]
+    assert compiled["plan_hash"] == bound_sql_plan_hash(compiled["plan"])
+    assert compiled["status"] == "compiled"
+
+
+def test_sql_compiler_hash_is_deterministic_and_identifiers_are_never_values():
+    first = _compile()
+    second = _compile()
+
+    # IDs are intentionally different in the helper, so pin the exact same
+    # governed inputs for a true determinism check.
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    input_schema, output_schema = customer_schemas()
+    input_binding = table_binding(input_schema, 'raw.customer')
+    output_binding = table_binding(
+        output_schema, 'clean.customer', access_mode="write"
+    )
+    rule = published_rule(
+        [{"id": "fill", "op": "fill_null", "column": "name", "value": "unknown"}]
+    )
+    compiler = SqlGlotRuleCompiler("postgresql")
+    kwargs = {
+        "rule_version": rule,
+        "input_schema": input_schema,
+        "output_schema": output_schema,
+        "input_binding": input_binding,
+        "output_binding": output_binding,
+        "backend": backend(),
+    }
+    deterministic = compiler.compile(**kwargs)
+    assert deterministic == compiler.compile(**copy.deepcopy(kwargs))
+    sql = deterministic["plan"]["statements"][0]["sql"]
+    assert "unknown" not in sql
+    assert deterministic["plan"]["statements"][0]["parameters"]
+    assert first["plan_hash"] != second["plan_hash"]
+
+
+def test_sql_compiler_fails_closed_for_unsupported_operations_and_binding_mismatch():
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    with pytest.raises(ValueError, match="unsupported SQL rule operation"):
+        _compile(
+            steps=[
+                {
+                    "id": "lookup",
+                    "op": "lookup_join",
+                    "lookup": {"schema_ref": "bd:restricted:v1"},
+                }
+            ]
+        )
+
+    with pytest.raises(ValueError, match="failure action"):
+        _compile(
+            steps=[
+                {
+                    "id": "quarantine",
+                    "op": "assert",
+                    "expression": "mobile != null",
+                    "on_failure": "quarantine",
+                    "severity": "error",
+                }
+            ]
+        )
+
+    input_schema, output_schema = customer_schemas()
+    input_binding = table_binding(input_schema, "raw.customer")
+    input_binding["schema_snapshot_id"] = new_governance_uid()
+    output_binding = table_binding(
+        output_schema, "clean.customer", access_mode="write"
+    )
+    rule = published_rule(
+        [{"id": "fill", "op": "fill_null", "column": "name", "value": "unknown"}]
+    )
+    with pytest.raises(ValueError, match="schema snapshot"):
+        SqlGlotRuleCompiler("postgresql").compile(
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=input_binding,
+            output_binding=output_binding,
+            backend=backend(),
+        )
+
+
+@pytest.mark.parametrize(
+    "step",
+    [
+        {
+            "id": "normalize_number",
+            "op": "normalize_text",
+            "column": "balance",
+            "trim": True,
+        },
+        {
+            "id": "bad_fill",
+            "op": "fill_null",
+            "column": "balance",
+            "value": "unknown",
+        },
+        {
+            "id": "bad_derive",
+            "op": "derive",
+            "target": "balance",
+            "expression": "mobile",
+        },
+        {
+            "id": "bad_map",
+            "op": "map_values",
+            "column": "balance",
+            "mapping": {"1": "2"},
+        },
+        {
+            "id": "bad_final_cast",
+            "op": "cast",
+            "column": "balance",
+            "to": "string",
+        },
+    ],
+)
+def test_sql_compiler_rejects_unproven_operator_type_semantics(step):
+    with pytest.raises(ValueError, match="type|string"):
+        _compile(steps=[step])
+
+
+@pytest.mark.parametrize(
+    "step",
+    [
+        {"id": "cast", "op": "cast", "column": "balance", "to": "decimal"},
+        {
+            "id": "normalize",
+            "op": "normalize_text",
+            "column": "name",
+            "trim": True,
+        },
+        {
+            "id": "regex",
+            "op": "regex_replace",
+            "column": "mobile",
+            "pattern": "[ -]",
+            "replacement": "_",
+        },
+        {"id": "fill", "op": "fill_null", "column": "name", "value": "unknown"},
+        {
+            "id": "filter",
+            "op": "filter",
+            "expression": "balance >= 0.00",
+        },
+        {
+            "id": "derive",
+            "op": "derive",
+            "target": "balance",
+            "expression": "round(balance, 2)",
+        },
+        {
+            "id": "assert",
+            "op": "assert",
+            "expression": "mobile != null",
+            "on_failure": "reject",
+            "severity": "error",
+        },
+        {
+            "id": "deduplicate",
+            "op": "deduplicate",
+            "keys": ["customer_id"],
+            "order_by": ["customer_id"],
+            "keep": "first",
+        },
+        {
+            "id": "map",
+            "op": "map_values",
+            "column": "name",
+            "mapping": {"A": "Alpha", "B": "Beta"},
+        },
+    ],
+)
+def test_sql_compiler_supports_only_provable_first_slice_operators(step):
+    compiled = _compile(steps=[step])
+
+    statement = compiled["plan"]["statements"][0]
+    assert re.match(r"^INSERT\b", statement["sql"], re.IGNORECASE)
+
+
+def test_compiler_registry_selects_only_same_source_concrete_sql_backends():
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+
+    input_schema, output_schema = customer_schemas()
+    source = table_binding(input_schema, "raw.customer")
+    target = table_binding(output_schema, "clean.customer", access_mode="write")
+    registry = CompilerRegistry(
+        {"postgresql": SqlGlotRuleCompiler("postgresql")}
+    )
+
+    selected = registry.select(
+        published_rule(
+            [{"id": "fill", "op": "fill_null", "column": "name", "value": "x"}]
+        )["rule_spec"],
+        source,
+        target,
+    )
+
+    assert isinstance(selected, SqlGlotRuleCompiler)
+    cross_source = copy.deepcopy(target)
+    cross_source["data_source_uid"] = new_governance_uid()
+    with pytest.raises(ValueError, match="cross-source"):
+        registry.select(
+            published_rule(
+                [
+                    {
+                        "id": "fill",
+                        "op": "fill_null",
+                        "column": "name",
+                        "value": "x",
+                    }
+                ]
+            )["rule_spec"],
+            source,
+            cross_source,
+        )
+
+
+def test_bound_plan_service_persists_compiled_state_without_claiming_publication():
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+    from app.core.data_rules.release import BoundSqlPlanService
+
+    input_schema, output_schema = customer_schemas()
+    source = table_binding(input_schema, "raw.customer")
+    target = table_binding(output_schema, "clean.customer", access_mode="write")
+    rule = published_rule(
+        [{"id": "fill", "op": "fill_null", "column": "name", "value": "x"}]
+    )
+
+    class Repository:
+        def __init__(self):
+            self.persisted = None
+
+        def persist_bound_component_plan(self, **kwargs):
+            self.persisted = kwargs
+            return {"id": new_governance_uid(), "status": kwargs["status"]}
+
+    repository = Repository()
+    result = BoundSqlPlanService(
+        repository,
+        CompilerRegistry({"postgresql": SqlGlotRuleCompiler("postgresql")}),
+    ).compile_and_persist(
+        component_binding_id=new_governance_uid(),
+        rule_version=rule,
+        input_schema=input_schema,
+        output_schema=output_schema,
+        input_binding=source,
+        output_binding=target,
+        backend=backend(),
+    )
+
+    assert result["status"] == "compiled"
+    assert repository.persisted["status"] == "compiled"
+    assert repository.persisted["compiled"]["plan"]["input_binding_id"] == source["id"]
+
+
+def test_repository_persists_bound_plan_only_for_one_deployment_linkage():
+    from app.core.data_rules.repository import DataRuleRepository
+
+    compiled = _compile()
+    component_binding_id = new_governance_uid()
+
+    class Mappings:
+        def __init__(self, row):
+            self.row = row
+
+        def one_or_none(self):
+            return self.row
+
+        def one(self):
+            return self.row
+
+    class Result:
+        def __init__(self, row):
+            self.row = row
+
+        def mappings(self):
+            return Mappings(self.row)
+
+    class Session:
+        def __init__(self, target_object_ref="clean.customer"):
+            self.calls = []
+            self.target_object_ref = target_object_ref
+
+        def execute(self, statement, params=None):
+            sql = str(statement)
+            values = params or {}
+            self.calls.append((sql, values))
+            if "bound_plan_linkage" in sql:
+                return Result(
+                    {
+                        "component_binding_id": component_binding_id,
+                        "input_schema_hash": "a" * 64,
+                        "output_schema_hash": "b" * 64,
+                        "input_object_ref": "raw.customer",
+                        "output_object_ref": self.target_object_ref,
+                        "data_source_uid": compiled["plan"]["data_source_uid"],
+                        "input_dialect": "postgresql",
+                        "output_dialect": "postgresql",
+                    }
+                )
+            if "INSERT INTO public.rule_execution_plans" in sql:
+                return Result(
+                    {
+                        "id": new_governance_uid(),
+                        "status": "compiled",
+                    }
+                )
+            return Result(None)
+
+    session = Session()
+    result = DataRuleRepository(session).persist_bound_component_plan(
+        component_binding_id=component_binding_id,
+        rule_version_id=compiled["plan"]["rule_version_id"],
+        input_binding_id=compiled["plan"]["input_binding_id"],
+        output_binding_id=compiled["plan"]["output_binding_id"],
+        compiled=compiled,
+        status="compiled",
+    )
+
+    assert result["status"] == "compiled"
+    insert = next(
+        params
+        for statement, params in session.calls
+        if "INSERT INTO public.rule_execution_plans" in statement
+    )
+    assert insert["plan_hash"] == compiled["plan_hash"]
+    assert insert["status"] == "compiled"
+    with pytest.raises(ValueError, match="compiled"):
+        DataRuleRepository(session).persist_bound_component_plan(
+            component_binding_id=component_binding_id,
+            rule_version_id=compiled["plan"]["rule_version_id"],
+            input_binding_id=compiled["plan"]["input_binding_id"],
+            output_binding_id=compiled["plan"]["output_binding_id"],
+            compiled=compiled,
+            status="published",
+        )
+
+    with pytest.raises(ValueError, match="physical dataset"):
+        DataRuleRepository(
+            Session(target_object_ref="clean.other_customer")
+        ).persist_bound_component_plan(
+            component_binding_id=component_binding_id,
+            rule_version_id=compiled["plan"]["rule_version_id"],
+            input_binding_id=compiled["plan"]["input_binding_id"],
+            output_binding_id=compiled["plan"]["output_binding_id"],
+            compiled=compiled,
+            status="compiled",
+        )

tests/integration/test_data_rule_sql_execution.py

@@ -0,0 +1,316 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+
+import pytest
+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
+from app.runner.nodes import NodeExecutionError
+
+CASES = [
+    (
+        "postgresql",
+        "postgresql+psycopg2://source_reader:source-test-password@127.0.0.1:25432/acceptance",
+        "public",
+        "C",
+        "posix",
+    ),
+    (
+        "mysql",
+        "mysql+pymysql://source_reader:source-test-password@127.0.0.1:23306/acceptance",
+        "acceptance",
+        "utf8mb4_0900_bin",
+        "icu",
+    ),
+]
+
+
+class Definition:
+    def __init__(self, dialect, capabilities):
+        self.database_type = dialect
+        self.extra_properties = {"sql_rule_capabilities": capabilities}
+
+
+class Definitions:
+    def __init__(self, definition):
+        self.definition = definition
+
+    def get(self, _uid):
+        return self.definition
+
+
+class DirectManager:
+    def __init__(self, engine, definition):
+        self.engine = engine
+        self.definitions = Definitions(definition)
+
+    @contextmanager
+    def connect(self, _uid, purpose):
+        assert purpose == "dataflow_write"
+        with self.engine.connect() as connection:
+            transaction = connection.begin()
+            try:
+                yield connection
+                transaction.commit()
+            except Exception:
+                transaction.rollback()
+                raise
+
+
+class PlanRepository:
+    def __init__(self, idempotency):
+        self.idempotency = idempotency
+        self.record = None
+
+    def persist_bound_component_plan(self, **kwargs):
+        compiled = kwargs["compiled"]
+        self.record = {
+            "component_binding_id": kwargs["component_binding_id"],
+            "rule_version_id": kwargs["rule_version_id"],
+            "backend": compiled["backend"],
+            "plan": compiled["plan"],
+            "plan_hash": compiled["plan_hash"],
+            "plan_status": kwargs["status"],
+            "rule_status": "published",
+            "component_kind": "rule.apply",
+            "binding_idempotency": self.idempotency,
+        }
+        return {
+            "id": new_governance_uid(),
+            "status": kwargs["status"],
+            "plan_hash": compiled["plan_hash"],
+        }
+
+    def publish_with_evidence(self, plan_hash, evidence):
+        assert self.record is not None
+        assert self.record["plan_status"] == "compiled"
+        assert self.record["plan_hash"] == plan_hash
+        assert evidence["commit_outcome"] == "committed"
+        assert evidence["rows_in"] >= evidence["rows_out"]
+        self.record["plan_status"] = "published"
+
+    def load(self, **_kwargs):
+        return dict(self.record)
+
+
+def _snapshot(schema_ref):
+    fields = [
+        {"name": "customer_id", "type": "integer", "nullable": False},
+        {"name": "name", "type": "string", "nullable": True},
+        {"name": "mobile", "type": "string", "nullable": True},
+    ]
+    return {
+        "id": new_governance_uid(),
+        "schema_ref": schema_ref,
+        "schema_hash": canonical_schema_hash(fields),
+        "fields": fields,
+        "source_revision": "task4:integration",
+    }
+
+
+@pytest.mark.parametrize(
+    ("dialect", "url", "schema_name", "collation", "regex_engine"), CASES
+)
+def test_bound_rule_compiles_publishes_executes_and_rejects_tampering(
+    dialect, url, schema_name, collation, regex_engine
+):
+    from app.core.data_rules.compilers import CompilerRegistry
+    from app.core.data_rules.compilers.sql import SqlGlotRuleCompiler
+    from app.core.data_rules.release import BoundSqlPlanService
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+    from app.runner.rules import RulePlanExecutor
+
+    engine = create_engine(url, pool_pre_ping=True)
+    source_name = "task4_rule_source"
+    target_name = "task4_rule_target"
+    source_ref = f"{schema_name}.{source_name}"
+    target_ref = f"{schema_name}.{target_name}"
+    capabilities = {
+        "dialect": dialect,
+        "timezone": "Asia/Shanghai",
+        "collation": collation,
+        "rounding_mode": "half_away_from_zero",
+        "regex_engine": regex_engine,
+    }
+    datasource_uid = new_governance_uid()
+    input_schema = _snapshot("bd:task4:raw")
+    output_schema = _snapshot("bd:task4:clean")
+    input_binding = {
+        "id": new_governance_uid(),
+        "data_source_uid": datasource_uid,
+        "object_kind": "table",
+        "object_ref": source_ref,
+        "schema_snapshot_id": input_schema["id"],
+        "access_mode": "read",
+        "dialect": dialect,
+        "write_mode": "append",
+    }
+    output_binding = {
+        "id": new_governance_uid(),
+        "data_source_uid": datasource_uid,
+        "object_kind": "table",
+        "object_ref": target_ref,
+        "schema_snapshot_id": output_schema["id"],
+        "access_mode": "write",
+        "dialect": dialect,
+        "write_mode": "append",
+    }
+    spec = validate_rule_spec(
+        {
+            "schema_version": "2.0",
+            "rule_uid": new_governance_uid(),
+            "name": "task4_real_sql",
+            "input_schema_ref": input_schema["schema_ref"],
+            "output_schema_ref": output_schema["schema_ref"],
+            "steps": [
+                {
+                    "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",
+                },
+            ],
+            "null_policy": "explicit",
+            "timezone": "Asia/Shanghai",
+        }
+    )
+    rule = {
+        "id": new_governance_uid(),
+        "status": "published",
+        "rule_spec": spec,
+        "spec_hash": rule_spec_hash(spec),
+    }
+
+    try:
+        with engine.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {target_name}"))
+            connection.execute(text(f"DROP TABLE IF EXISTS {source_name}"))
+            connection.execute(
+                text(
+                    f"CREATE TABLE {source_name} ("
+                    "customer_id BIGINT PRIMARY KEY, "
+                    "name VARCHAR(100), mobile VARCHAR(30))"
+                )
+            )
+            connection.execute(
+                text(
+                    f"CREATE TABLE {target_name} ("
+                    "customer_id BIGINT PRIMARY KEY, "
+                    "name VARCHAR(100), mobile VARCHAR(30))"
+                )
+            )
+            connection.execute(
+                text(
+                    f"INSERT INTO {source_name} "
+                    "(customer_id, name, mobile) VALUES "
+                    "(1, ' Alice ', '13800138000'), "
+                    "(2, ' Bad ', 'not-a-mobile')"
+                )
+            )
+
+        component_binding_id = new_governance_uid()
+        idempotency = {
+            "strategy": "upsert",
+            "key": "customer_id",
+        }
+        repository = PlanRepository(idempotency)
+        BoundSqlPlanService(
+            repository,
+            CompilerRegistry(
+                {dialect: SqlGlotRuleCompiler(dialect)}
+            ),
+        ).compile_and_persist(
+            component_binding_id=component_binding_id,
+            rule_version=rule,
+            input_schema=input_schema,
+            output_schema=output_schema,
+            input_binding=input_binding,
+            output_binding=output_binding,
+            backend=capabilities,
+        )
+        record = repository.record
+        assert record["plan_status"] == "compiled"
+        compiled = {
+            "plan": record["plan"],
+            "plan_hash": record["plan_hash"],
+        }
+        adapter = SqlGlotRulePlanAdapter(
+            DirectManager(
+                engine,
+                Definition(dialect, capabilities),
+            )
+        )
+        node = {
+            "id": "task4_real_rule",
+            "type": "rule.apply",
+            "purpose": "write",
+            "idempotency": idempotency,
+            "config": {
+                "component_binding_id": component_binding_id,
+                "rule_version_id": rule["id"],
+                "execution_plan_hash": compiled["plan_hash"],
+            },
+        }
+        preflight_evidence = adapter.execute(
+            plan=compiled["plan"],
+            node=node,
+            parameters={},
+            write_authorized=True,
+        )
+        with engine.begin() as connection:
+            connection.execute(text(f"DELETE FROM {target_name}"))
+        repository.publish_with_evidence(
+            compiled["plan_hash"],
+            preflight_evidence,
+        )
+        executor = RulePlanExecutor(
+            repository,
+            adapters={"sql_pushdown": adapter},
+        )
+
+        result = executor.execute(node, {}, write_authorized=True)
+
+        assert result["rows_in"] == 2
+        assert result["rows_out"] == 1
+        assert result["rows_rejected"] == 1
+        with engine.connect() as connection:
+            rows = connection.execute(
+                text(
+                    f"SELECT customer_id, name, mobile "
+                    f"FROM {target_name} ORDER BY customer_id"
+                )
+            ).tuples().all()
+        assert rows == [(1, "Alice", "13800138000")]
+
+        repeated = executor.execute(node, {}, write_authorized=True)
+        assert repeated["rows_out"] == 1
+        assert repeated["rows_rejected"] == 1
+        with engine.connect() as connection:
+            assert connection.execute(
+                text(f"SELECT COUNT(*) FROM {target_name}")
+            ).scalar_one() == 1
+
+        repository.record["plan"] = {
+            **repository.record["plan"],
+            "result_contract": {
+                **repository.record["plan"]["result_contract"],
+                "rows_rejected": "unknown",
+            },
+        }
+        with pytest.raises(NodeExecutionError, match="not executable"):
+            executor.execute(node, {}, write_authorized=True)
+    finally:
+        with engine.begin() as connection:
+            connection.execute(text(f"DROP TABLE IF EXISTS {target_name}"))
+            connection.execute(text(f"DROP TABLE IF EXISTS {source_name}"))
+        engine.dispose()

tests/runner/test_rule_sql.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+
+import pytest
+
+from app.core.common.identifiers import new_governance_uid
+from app.core.data_source.errors import DataSourceWriteOutcomeUnknown
+from app.runner.nodes import NodeExecutionError
+
+
+class Definition:
+    def __init__(self, dialect):
+        self.database_type = dialect
+        self.extra_properties = {
+            "sql_rule_capabilities": {
+                "dialect": dialect,
+                "timezone": "Asia/Shanghai",
+                "collation": "C" if dialect == "postgresql" else "utf8mb4_0900_bin",
+                "rounding_mode": "half_away_from_zero",
+                "regex_engine": "posix" if dialect == "postgresql" else "icu",
+            }
+        }
+
+
+class Definitions:
+    def __init__(self, definition):
+        self.definition = definition
+
+    def get(self, _uid):
+        return self.definition
+
+
+class Result:
+    def __init__(self, *, scalar=None, rowcount=0):
+        self._scalar = scalar
+        self.rowcount = rowcount
+
+    def scalar_one(self):
+        return self._scalar
+
+
+class Connection:
+    def __init__(self):
+        self.calls = []
+
+    def execute(self, statement, parameters):
+        self.calls.append((str(statement), parameters))
+        if len(self.calls) == 1:
+            return Result(scalar=3)
+        if len(self.calls) == 2:
+            return Result(scalar=2)
+        return Result(rowcount=2)
+
+
+class Manager:
+    def __init__(self, dialect="postgresql", *, unknown_commit=False):
+        self.definitions = Definitions(Definition(dialect))
+        self.connection = Connection()
+        self.unknown_commit = unknown_commit
+        self.calls = []
+
+    @contextmanager
+    def connect(self, uid, purpose):
+        self.calls.append((uid, purpose))
+        yield self.connection
+        if self.unknown_commit:
+            raise DataSourceWriteOutcomeUnknown()
+
+
+def sql_plan(dialect="postgresql"):
+    from app.core.data_rules.compilers.sql import bound_sql_plan_hash
+
+    uid = new_governance_uid()
+    capabilities = Definition(dialect).extra_properties["sql_rule_capabilities"]
+    quote = '"' if dialect == "postgresql" else "`"
+    plan = {
+        "schema_version": "1.0",
+        "dialect": dialect,
+        "capabilities": capabilities,
+        "data_source_uid": uid,
+        "rule_version_id": new_governance_uid(),
+        "input_binding_id": new_governance_uid(),
+        "output_binding_id": new_governance_uid(),
+        "statements": [
+            {
+                "purpose": "write",
+                "sql": (
+                    f"INSERT INTO {quote}clean{quote}.{quote}customer{quote} "
+                    f"({quote}id{quote}) SELECT {quote}id{quote} "
+                    f"FROM {quote}raw{quote}.{quote}customer{quote}"
+                ),
+                "parameters": {},
+            }
+        ],
+        "result_contract": {
+            "rows_in": "counted",
+            "rows_out": "counted",
+            "rows_rejected": "counted",
+        },
+    }
+    return plan, bound_sql_plan_hash(plan)
+
+
+def node_for(plan, plan_hash):
+    return {
+        "id": "task4_rule",
+        "type": "rule.apply",
+        "purpose": "write",
+        "idempotency": {"strategy": "upsert", "key": "id"},
+        "config": {
+            "component_binding_id": new_governance_uid(),
+            "rule_version_id": plan["rule_version_id"],
+            "execution_plan_hash": plan_hash,
+        },
+    }
+
+
+def test_sqlglot_rule_adapter_verifies_and_executes_one_transaction():
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, plan_hash = sql_plan()
+    node = node_for(plan, plan_hash)
+    manager = Manager()
+
+    result = SqlGlotRulePlanAdapter(manager).execute(
+        plan=plan,
+        node=node,
+        parameters={},
+        write_authorized=True,
+    )
+
+    assert result == {
+        "rows_in": 3,
+        "rows_out": 2,
+        "rows_rejected": 1,
+        "commit_outcome": "committed",
+    }
+    assert manager.calls == [(plan["data_source_uid"], "dataflow_write")]
+    assert len(manager.connection.calls) == 3
+
+
+def test_sqlglot_rule_adapter_fails_closed_for_dialect_hash_and_authorization():
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, plan_hash = sql_plan()
+    node = node_for(plan, plan_hash)
+
+    with pytest.raises(NodeExecutionError, match="dialect"):
+        SqlGlotRulePlanAdapter(Manager("mysql")).execute(
+            plan=plan,
+            node=node,
+            parameters={},
+            write_authorized=True,
+        )
+
+    node["config"]["execution_plan_hash"] = "0" * 64
+    with pytest.raises(NodeExecutionError, match="hash"):
+        SqlGlotRulePlanAdapter(Manager()).execute(
+            plan=plan,
+            node=node,
+            parameters={},
+            write_authorized=True,
+        )
+
+    node["config"]["execution_plan_hash"] = plan_hash
+    with pytest.raises(NodeExecutionError, match="authorization"):
+        SqlGlotRulePlanAdapter(Manager()).execute(
+            plan=plan,
+            node=node,
+            parameters={},
+            write_authorized=False,
+        )
+
+
+def test_sqlglot_rule_adapter_rejects_unimplemented_idempotency_strategy():
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, plan_hash = sql_plan()
+    node = node_for(plan, plan_hash)
+    node["idempotency"] = {
+        "strategy": "partition_replace",
+        "key": "task4",
+    }
+
+    with pytest.raises(NodeExecutionError, match="idempotency"):
+        SqlGlotRulePlanAdapter(Manager()).execute(
+            plan=plan,
+            node=node,
+            parameters={},
+            write_authorized=True,
+        )
+
+
+def test_rule_executor_matches_node_idempotency_to_persisted_component():
+    from app.runner.rules import RulePlanExecutor
+
+    plan, plan_hash = sql_plan()
+    node = node_for(plan, plan_hash)
+
+    class Repository:
+        def load(self, **_kwargs):
+            return {
+                "component_binding_id": node["config"]["component_binding_id"],
+                "rule_version_id": plan["rule_version_id"],
+                "backend": "sql_pushdown",
+                "plan": plan,
+                "plan_hash": plan_hash,
+                "plan_status": "published",
+                "rule_status": "published",
+                "component_kind": "rule.apply",
+                "binding_idempotency": {
+                    "strategy": "upsert",
+                    "key": "different_id",
+                },
+            }
+
+    class Adapter:
+        def execute(self, **_kwargs):
+            return {"rows_in": 0, "rows_out": 0, "rows_rejected": 0}
+
+    with pytest.raises(NodeExecutionError, match="idempotency"):
+        RulePlanExecutor(
+            Repository(),
+            adapters={"sql_pushdown": Adapter()},
+        ).execute(node, {}, write_authorized=True)
+
+
+def test_sqlglot_rule_adapter_reports_unknown_commit_outcome():
+    from app.runner.rule_sql import SqlGlotRulePlanAdapter
+
+    plan, plan_hash = sql_plan()
+
+    with pytest.raises(NodeExecutionError) as error:
+        SqlGlotRulePlanAdapter(Manager(unknown_commit=True)).execute(
+            plan=plan,
+            node=node_for(plan, plan_hash),
+            parameters={},
+            write_authorized=True,
+        )
+
+    assert error.value.commit_outcome == "unknown"

tests/runner/test_rules.py

@@ -9,7 +9,6 @@ import pytest
 from app.core.common.identifiers import new_governance_uid
 from app.runner.nodes import NodeExecutionError
 
-
 PLAN = {"op": "not_null", "column": "mobile"}
 PLAN_HASH = hashlib.sha256(
     json.dumps(
@@ -75,6 +74,8 @@ def published_record(node, **overrides):
         "plan_hash": PLAN_HASH,
         "plan_status": "published",
         "rule_status": "published",
+        "component_kind": node["type"],
+        "binding_idempotency": node.get("idempotency"),
     }
     value.update(overrides)
     return value