govoplan-dataflow/src/govoplan_dataflow/backend/executor.py

from __future__ import annotations

import hashlib
import json
import time
from collections import defaultdict
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any, Callable, NoReturn

from govoplan_dataflow.backend.expressions import (
    convert_value,
    evaluate_expression,
    parse_expression,
)
from govoplan_dataflow.backend.graph import graph_inputs_by_port, topological_order, validate_graph
from govoplan_dataflow.backend.operator_registry import (
    OPERATOR_REGISTRY,
    OperatorExecutionContext,
    OperatorExecutionResult,
)
from govoplan_dataflow.backend.preview_limits import MAX_RESULT_BYTES
from govoplan_dataflow.backend.schemas import (
    DataflowDiagnostic,
    GraphNode,
    NodePreviewDiagnostic,
    NodePreviewResult,
    PipelineGraph,
    PreviewColumn,
)
from govoplan_dataflow.backend.subflows import substitute_parameters


EXECUTOR_VERSION = "dataflow-preview-v2"
MAX_EXECUTION_SECONDS = 2.0
MAX_SOURCE_ROWS = 250
MAX_INTERMEDIATE_ROWS = 10_000


class PipelineExecutionError(RuntimeError):
    def __init__(
        self,
        message: str,
        *,
        node_id: str | None = None,
        node_diagnostics: tuple[NodePreviewDiagnostic, ...] = (),
        source_fingerprints: tuple[dict[str, Any], ...] = (),
        input_row_count: int = 0,
        diagnostics: tuple[DataflowDiagnostic, ...] = (),
        node_preview: NodePreviewResult | None = None,
        retryable: bool = False,
    ) -> None:
        super().__init__(message)
        self.node_id = node_id
        self.node_diagnostics = node_diagnostics
        self.source_fingerprints = source_fingerprints
        self.input_row_count = input_row_count
        self.diagnostics = diagnostics
        self.node_preview = node_preview
        self.retryable = retryable


@dataclass(frozen=True)
class PipelineExecutionResult:
    rows: list[dict[str, Any]]
    total_rows: int
    truncated: bool
    columns: list[PreviewColumn]
    diagnostics: list[DataflowDiagnostic]
    node_diagnostics: list[NodePreviewDiagnostic]
    node_preview: NodePreviewResult | None
    source_fingerprints: list[dict[str, Any]]
    input_row_count: int


@dataclass(frozen=True)
class ResolvedSource:
    rows: tuple[dict[str, Any], ...]
    source_ref: str
    provider: str
    fingerprint: str
    total_rows: int
    truncated: bool = False


SourceResolver = Callable[[GraphNode, int], ResolvedSource]


@dataclass(frozen=True, slots=True)
class _PreviewPlan:
    graph: PipelineGraph
    node_by_id: dict[str, GraphNode]
    inputs_by_port: dict[str, dict[str, list[str]]]
    ordered_node_ids: list[str]
    validation: list[DataflowDiagnostic]
    preview_node_id: str | None
    row_limit: int


@dataclass(slots=True)
class _PreviewState:
    started: float = field(default_factory=time.monotonic)
    outputs: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
    node_diagnostics: list[NodePreviewDiagnostic] = field(default_factory=list)
    source_fingerprints: list[dict[str, Any]] = field(default_factory=list)
    input_row_count: int = 0


def execute_preview(
    graph: PipelineGraph,
    *,
    row_limit: int,
    source_resolver: SourceResolver | None = None,
    preview_node_id: str | None = None,
    _execution_depth: int = 0,
) -> PipelineExecutionResult:
    plan = _prepare_preview(
        graph,
        row_limit=row_limit,
        preview_node_id=preview_node_id,
        execution_depth=_execution_depth,
    )
    state = _PreviewState()
    for node_id in plan.ordered_node_ids:
        _execute_preview_node(
            plan,
            state,
            plan.node_by_id[node_id],
            source_resolver=source_resolver,
            execution_depth=_execution_depth,
        )
    return _preview_result(plan, state)


def _prepare_preview(
    graph: PipelineGraph,
    *,
    row_limit: int,
    preview_node_id: str | None,
    execution_depth: int,
) -> _PreviewPlan:
    if execution_depth > 5:
        raise PipelineExecutionError("Subflows are limited to five nested levels.")
    validation = validate_graph(graph)
    first_error = next(
        (item for item in validation if item.severity == "error"),
        None,
    )
    if first_error is not None:
        raise PipelineExecutionError(
            first_error.message,
            node_id=first_error.node_id,
        )
    node_by_id = {node.id: node for node in graph.nodes}
    if preview_node_id is not None and preview_node_id not in node_by_id:
        raise PipelineExecutionError(
            "The requested preview node is not part of this pipeline.",
            node_id=preview_node_id,
        )
    ordered, cyclic = topological_order(graph)
    if cyclic:
        raise PipelineExecutionError("Pipeline graph contains a cycle")
    return _PreviewPlan(
        graph=graph,
        node_by_id=node_by_id,
        inputs_by_port=graph_inputs_by_port(graph),
        ordered_node_ids=ordered,
        validation=validation,
        preview_node_id=preview_node_id,
        row_limit=row_limit,
    )


def _execute_preview_node(
    plan: _PreviewPlan,
    state: _PreviewState,
    node: GraphNode,
    *,
    source_resolver: SourceResolver | None,
    execution_depth: int,
) -> None:
    node_started = time.monotonic()
    node_inputs = plan.inputs_by_port.get(node.id, {})
    input_sets = [
        state.outputs[source_id]
        for port_sources in node_inputs.values()
        for source_id in port_sources
    ]
    _enforce_preview_deadline(plan, state, node, input_sets)
    try:
        execution = _run_operator(
            node,
            node_inputs=node_inputs,
            input_sets=input_sets,
            outputs=state.outputs,
            source_resolver=source_resolver,
            execution_depth=execution_depth,
        )
        _record_execution(plan, state, node, execution)
        _guard_result_size(execution.rows, node_id=node.id)
    except (
        PipelineExecutionError,
        ArithmeticError,
        KeyError,
        TypeError,
        ValueError,
    ) as exc:
        _raise_preview_node_failure(
            plan,
            state,
            node,
            input_sets=input_sets,
            node_started=node_started,
            exc=exc,
        )
    state.outputs[node.id] = execution.rows
    state.node_diagnostics.append(
        _successful_node_diagnostic(
            node,
            execution,
            input_sets=input_sets,
            node_started=node_started,
        )
    )



def _run_operator(
    node: GraphNode,
    *,
    node_inputs: dict[str, list[str]],
    input_sets: list[list[dict[str, Any]]],
    outputs: dict[str, list[dict[str, Any]]],
    source_resolver: SourceResolver | None,
    execution_depth: int,
) -> OperatorExecutionResult:
    executor = OPERATOR_REGISTRY.executor(node.type)
    if executor is None:
        raise PipelineExecutionError(
            f"Node type {node.type!r} has no registered executor.",
            node_id=node.id,
        )
    return executor(
        OperatorExecutionContext(
            node=node,
            inputs_by_port=node_inputs,
            outputs=outputs,
            input_sets=input_sets,
            input_rows=input_sets[0] if len(input_sets) == 1 else [],
            source_resolver=source_resolver,
            execution_depth=execution_depth,
        )
    )


def _record_execution(
    plan: _PreviewPlan,
    state: _PreviewState,
    node: GraphNode,
    execution: OperatorExecutionResult,
) -> None:
    state.input_row_count += execution.input_row_count
    state.source_fingerprints.extend(execution.source_fingerprints)
    if node.type == "source.reference":
        plan.validation.extend(
            DataflowDiagnostic(
                severity="warning",
                code="source.preview_truncated",
                message=message,
                node_id=node.id,
            )
            for message in execution.messages
        )


def _guard_result_size(
    rows: list[dict[str, Any]],
    *,
    node_id: str,
) -> None:
    if len(json.dumps(rows, default=str).encode("utf-8")) <= MAX_RESULT_BYTES:
        return
    raise PipelineExecutionError(
        "A preview node exceeded the one-megabyte result limit.",
        node_id=node_id,
    )


def _enforce_preview_deadline(
    plan: _PreviewPlan,
    state: _PreviewState,
    node: GraphNode,
    input_sets: list[list[dict[str, Any]]],
) -> None:
    if time.monotonic() - state.started <= MAX_EXECUTION_SECONDS:
        return
    message = "Preview exceeded the two-second execution limit"
    failed = NodePreviewDiagnostic(
        node_id=node.id,
        status="failed",
        input_rows=_input_row_total(input_sets),
        output_rows=0,
        duration_ms=0,
        columns=[],
        messages=[message],
    )
    raise PipelineExecutionError(
        message,
        node_id=node.id,
        node_diagnostics=tuple([*state.node_diagnostics, failed]),
        source_fingerprints=tuple(state.source_fingerprints),
        input_row_count=state.input_row_count,
        diagnostics=tuple(_warnings(plan.validation)),
        node_preview=_node_preview(
            state.outputs,
            plan.preview_node_id,
            plan.row_limit,
        ),
    )


def _raise_preview_node_failure(
    plan: _PreviewPlan,
    state: _PreviewState,
    node: GraphNode,
    *,
    input_sets: list[list[dict[str, Any]]],
    node_started: float,
    exc: BaseException,
) -> NoReturn:
    execution_error = (
        exc
        if isinstance(exc, PipelineExecutionError)
        else PipelineExecutionError(str(exc), node_id=node.id)
    )
    failed = NodePreviewDiagnostic(
        node_id=execution_error.node_id or node.id,
        status="failed",
        input_rows=_input_row_total(input_sets),
        output_rows=0,
        duration_ms=_elapsed_ms(node_started),
        columns=[],
        messages=[str(execution_error)],
    )
    raise PipelineExecutionError(
        str(execution_error),
        node_id=execution_error.node_id or node.id,
        node_diagnostics=tuple([*state.node_diagnostics, failed]),
        source_fingerprints=tuple(state.source_fingerprints),
        input_row_count=state.input_row_count,
        diagnostics=tuple(_warnings(plan.validation)),
        node_preview=_node_preview(
            state.outputs,
            plan.preview_node_id,
            plan.row_limit,
        ),
    ) from exc


def _successful_node_diagnostic(
    node: GraphNode,
    execution: OperatorExecutionResult,
    *,
    input_sets: list[list[dict[str, Any]]],
    node_started: float,
) -> NodePreviewDiagnostic:
    return NodePreviewDiagnostic(
        node_id=node.id,
        status="succeeded",
        input_rows=_input_row_total(input_sets),
        output_rows=len(execution.rows),
        duration_ms=_elapsed_ms(node_started),
        columns=infer_columns(execution.rows),
        messages=list(execution.messages),
    )


def _preview_result(
    plan: _PreviewPlan,
    state: _PreviewState,
) -> PipelineExecutionResult:
    output_node = next(
        node
        for node in plan.graph.nodes
        if node.type == "output"
    )
    all_rows = state.outputs[output_node.id]
    rows = all_rows[: plan.row_limit]
    return PipelineExecutionResult(
        rows=rows,
        total_rows=len(all_rows),
        truncated=len(rows) < len(all_rows),
        columns=infer_columns(all_rows),
        diagnostics=_warnings(plan.validation),
        node_diagnostics=state.node_diagnostics,
        node_preview=_node_preview(
            state.outputs,
            plan.preview_node_id,
            plan.row_limit,
        ),
        source_fingerprints=state.source_fingerprints,
        input_row_count=state.input_row_count,
    )


def _warnings(
    diagnostics: list[DataflowDiagnostic],
) -> list[DataflowDiagnostic]:
    return [
        item
        for item in diagnostics
        if item.severity != "error"
    ]


def _input_row_total(
    input_sets: list[list[dict[str, Any]]],
) -> int:
    return sum(len(rows) for rows in input_sets)


def _elapsed_ms(started: float) -> float:
    return round((time.monotonic() - started) * 1000, 3)


def _node_preview(
    outputs: dict[str, list[dict[str, Any]]],
    node_id: str | None,
    row_limit: int,
) -> NodePreviewResult | None:
    if node_id is None or node_id not in outputs:
        return None
    all_rows = outputs[node_id]
    rows = [dict(row) for row in all_rows[:row_limit]]
    return NodePreviewResult(
        node_id=node_id,
        columns=infer_columns(all_rows),
        rows=rows,
        total_rows=len(all_rows),
        truncated=len(rows) < len(all_rows),
    )


def infer_columns(rows: list[dict[str, Any]]) -> list[PreviewColumn]:
    names: list[str] = []
    for row in rows:
        for name in row:
            if name not in names:
                names.append(name)
    columns: list[PreviewColumn] = []
    for name in names:
        values = [row.get(name) for row in rows]
        concrete = [value for value in values if value is not None]
        inferred = _type_name(concrete[0]) if concrete else "unknown"
        if any(_type_name(value) != inferred for value in concrete[1:]):
            inferred = "mixed"
        columns.append(
            PreviewColumn(
                name=name,
                type=inferred,
                nullable=len(concrete) != len(values),
            )
        )
    return columns


def _union_rows(
    inputs: list[list[dict[str, Any]]],
    config: dict[str, Any],
) -> list[dict[str, Any]]:
    rows = [dict(row) for input_rows in inputs for row in input_rows]
    if config.get("mode", "all") == "distinct":
        return _distinct_rows(rows, {})
    return rows


def _join_rows(
    left_rows: list[dict[str, Any]],
    right_rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    left_keys = [str(item) for item in config["left_keys"]]
    right_keys = [str(item) for item in config["right_keys"]]
    join_type = str(config.get("join_type", "inner"))
    right_prefix = str(config.get("right_prefix", "right_"))
    left_columns = _ordered_columns(left_rows)
    right_columns = _ordered_columns(right_rows)
    right_index: dict[tuple[Any, ...], list[tuple[int, dict[str, Any]]]] = defaultdict(list)
    for index, row in enumerate(right_rows):
        key = _join_key(row, right_keys)
        if key is not None:
            right_index[key].append((index, row))

    output: list[dict[str, Any]] = []
    matched_right: set[int] = set()
    for left_row in left_rows:
        key = _join_key(left_row, left_keys)
        matches = right_index.get(key, ()) if key is not None else ()
        if join_type == "semi":
            if matches:
                output.append(dict(left_row))
                _guard_intermediate_size(output, node_id=node_id)
            continue
        if join_type == "anti":
            if not matches:
                output.append(dict(left_row))
                _guard_intermediate_size(output, node_id=node_id)
            continue
        if matches:
            for right_index_value, right_row in matches:
                matched_right.add(right_index_value)
                output.append(
                    _merge_join_rows(
                        left_row,
                        right_row,
                        left_columns=left_columns,
                        right_columns=right_columns,
                        right_prefix=right_prefix,
                    )
                )
                _guard_intermediate_size(output, node_id=node_id)
        elif join_type in {"left", "full"}:
            output.append(
                _merge_join_rows(
                    left_row,
                    None,
                    left_columns=left_columns,
                    right_columns=right_columns,
                    right_prefix=right_prefix,
                )
            )

    if join_type in {"right", "full"}:
        for index, right_row in enumerate(right_rows):
            if index in matched_right:
                continue
            output.append(
                _merge_join_rows(
                    None,
                    right_row,
                    left_columns=left_columns,
                    right_columns=right_columns,
                    right_prefix=right_prefix,
                )
            )
            _guard_intermediate_size(output, node_id=node_id)
    return output


def _distinct_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
) -> list[dict[str, Any]]:
    columns = [str(item) for item in config.get("columns", [])]
    seen: set[str] = set()
    output: list[dict[str, Any]] = []
    for row in rows:
        value = {column: row.get(column) for column in columns} if columns else row
        identity = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
        if identity in seen:
            continue
        seen.add(identity)
        output.append(dict(row))
    return output


def _derive_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    target = str(config["target_column"])
    operation = str(config["operation"])
    columns = [str(item) for item in config["source_columns"]]
    separator = str(config.get("separator", " "))
    output: list[dict[str, Any]] = []
    for row in rows:
        values = [row.get(column) for column in columns]
        try:
            derived = _derive_value(operation, values, separator=separator)
        except (ArithmeticError, TypeError, ValueError) as exc:
            raise PipelineExecutionError(
                f"Cannot derive {target!r} with {operation!r}: {exc}",
                node_id=node_id,
            ) from exc
        result = dict(row)
        result[target] = derived
        output.append(result)
    return output


def _derive_value(operation: str, values: list[Any], *, separator: str) -> Any:
    handler = _DERIVE_HANDLERS.get(operation)
    if handler is None:
        raise ValueError(f"unknown derive operation {operation!r}")
    return handler(values, separator)


def _unary_text(
    values: list[Any],
    _separator: str,
    operation: Callable[[str], str],
) -> str | None:
    return None if values[0] is None else operation(str(values[0]))


def _derive_concat(values: list[Any], separator: str) -> str:
    return separator.join(str(value) for value in values if value is not None)


def _derive_coalesce(values: list[Any], _separator: str) -> Any:
    return next((value for value in values if value not in (None, "")), None)


def _derive_numeric(
    values: list[Any],
    _separator: str,
    operation: Callable[[Any, Any], Any],
) -> Any:
    if any(value is None for value in values):
        return None
    left, right = values
    if isinstance(left, bool) or isinstance(right, bool):
        raise TypeError("boolean values are not numeric inputs")
    return operation(left, right)


_DERIVE_HANDLERS: dict[str, Callable[[list[Any], str], Any]] = {
    "copy": lambda values, _separator: values[0],
    "upper": lambda values, separator: _unary_text(
        values, separator, str.upper
    ),
    "lower": lambda values, separator: _unary_text(
        values, separator, str.lower
    ),
    "trim": lambda values, separator: _unary_text(
        values, separator, str.strip
    ),
    "concat": _derive_concat,
    "coalesce": _derive_coalesce,
    "add": lambda values, separator: _derive_numeric(
        values, separator, lambda left, right: left + right
    ),
    "subtract": lambda values, separator: _derive_numeric(
        values, separator, lambda left, right: left - right
    ),
    "multiply": lambda values, separator: _derive_numeric(
        values, separator, lambda left, right: left * right
    ),
    "divide": lambda values, separator: _derive_numeric(
        values, separator, lambda left, right: left / right
    ),
}


def _expression_filter_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    parsed = parse_expression(str(config["expression"]))
    result: list[dict[str, Any]] = []
    for row in rows:
        try:
            if bool(evaluate_expression(parsed, row)):
                result.append(dict(row))
        except (ArithmeticError, TypeError, ValueError) as exc:
            raise PipelineExecutionError(
                f"Cannot evaluate filter expression: {exc}",
                node_id=node_id,
            ) from exc
    return result


def _expression_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    target = str(config["target_column"])
    parsed = parse_expression(str(config["expression"]))
    output: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        try:
            item[target] = evaluate_expression(parsed, row)
        except (ArithmeticError, TypeError, ValueError) as exc:
            raise PipelineExecutionError(
                f"Cannot evaluate expression for {target!r}: {exc}",
                node_id=node_id,
            ) from exc
        output.append(item)
    return output


def _calculation_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    calculations: list[tuple[str, Any]] = []
    try:
        calculations = [
            (
                str(item["target_column"]),
                parse_expression(str(item["expression"])),
            )
            for item in config["calculations"]
        ]
    except (KeyError, TypeError, ValueError) as exc:
        raise PipelineExecutionError(
            f"Cannot prepare calculated columns: {exc}",
            node_id=node_id,
        ) from exc
    output: list[dict[str, Any]] = []
    for row in rows:
        result = dict(row)
        for target, parsed in calculations:
            try:
                result[target] = evaluate_expression(parsed, result)
            except (ArithmeticError, TypeError, ValueError) as exc:
                raise PipelineExecutionError(
                    f"Cannot calculate {target!r}: {exc}",
                    node_id=node_id,
                ) from exc
        output.append(result)
    return output


def _convert_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    source = str(config["source_column"])
    target = str(config["target_column"])
    target_type = str(config["target_type"])
    on_error = str(config.get("on_error", "fail"))
    output: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        try:
            item[target] = convert_value(
                row.get(source),
                target_type,
                on_error=on_error,  # type: ignore[arg-type]
            )
        except (ArithmeticError, TypeError, ValueError) as exc:
            raise PipelineExecutionError(
                f"Cannot convert {source!r} to {target_type}: {exc}",
                node_id=node_id,
            ) from exc
        output.append(item)
    return output


def _replace_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
) -> list[dict[str, Any]]:
    source = str(config["source_column"])
    target = str(config["target_column"])
    mode = str(config.get("mode", "exact"))
    find = config.get("find")
    replacement = config.get("replacement")
    output: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        value = row.get(source)
        if mode == "text" and value is not None:
            item[target] = str(value).replace(str(find), str(replacement))
        else:
            item[target] = replacement if value == find else value
        output.append(item)
    return output


def _quality_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    rules = config.get("rules", [])
    action = str(config.get("action", "annotate"))
    unique_values: dict[str, set[object]] = defaultdict(set)
    output: list[dict[str, Any]] = []
    invalid_count = 0
    for row_index, row in enumerate(rows, start=1):
        failures: list[str] = []
        for rule in rules:
            if not isinstance(rule, dict):
                continue
            rule_id = str(rule.get("id") or rule.get("operator") or "rule")
            column = str(rule.get("column") or "")
            value = row.get(column)
            operator = str(rule.get("operator") or "")
            valid = _quality_rule_matches(
                value,
                operator=operator,
                rule=rule,
                seen=unique_values[rule_id],
            )
            if not valid:
                failures.append(rule_id)
        if failures:
            invalid_count += 1
            if action == "fail":
                raise PipelineExecutionError(
                    f"Quality rules failed for row {row_index}: {', '.join(failures)}.",
                    node_id=node_id,
                )
            if action == "drop":
                continue
        item = dict(row)
        if action == "annotate":
            item["_quality_valid"] = not failures
            item["_quality_errors"] = failures
        output.append(item)
    if invalid_count and action not in {"annotate", "drop", "fail"}:
        raise PipelineExecutionError(
            f"Unsupported quality action {action!r}.",
            node_id=node_id,
        )
    return output


def _quality_rule_matches(
    value: Any,
    *,
    operator: str,
    rule: dict[str, Any],
    seen: set[object],
) -> bool:
    if operator == "not_null":
        return value is not None and value != ""
    if operator == "type":
        expected = str(rule.get("value") or "")
        mapping = {
            "string": str,
            "integer": int,
            "number": (int, float, Decimal),
            "boolean": bool,
        }
        expected_type = mapping.get(expected)
        return expected_type is not None and isinstance(value, expected_type)
    if operator == "min":
        return value is not None and value >= rule.get("value")
    if operator == "max":
        return value is not None and value <= rule.get("value")
    if operator == "allowed":
        allowed = rule.get("values")
        return isinstance(allowed, list) and value in allowed
    if operator == "unique":
        marker = _hashable(value)
        if marker in seen:
            return False
        seen.add(marker)
        return True
    return False


def _reconcile_rows(
    left_rows: list[dict[str, Any]],
    right_rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    left_keys = [str(item) for item in config["left_keys"]]
    right_keys = [str(item) for item in config["right_keys"]]
    right_prefix = str(config.get("right_prefix", "observed_"))
    comparisons = _comparison_fields(config.get("compare_columns"))
    left_index = _unique_row_index(
        left_rows,
        left_keys,
        node_id=node_id,
        input_label="expected",
    )
    right_index = _unique_row_index(
        right_rows,
        right_keys,
        node_id=node_id,
        input_label="observed",
    )
    output: list[dict[str, Any]] = []
    for key in dict.fromkeys((*left_index, *right_index)):
        left = left_index.get(key)
        right = right_index.get(key)
        key_values = [
            (left or {}).get(left_name)
            if left is not None
            else (right or {}).get(right_name)
            for left_name, right_name in zip(
                left_keys,
                right_keys,
                strict=True,
            )
        ]
        item = dict(left or {})
        if right is not None:
            item.update({f"{right_prefix}{name}": value for name, value in right.items()})
        if left is None:
            status = "missing_expected"
            differences: list[str] = []
        elif right is None:
            status = "missing_observed"
            differences = []
        else:
            fields = comparisons or tuple((name, name) for name in left if name not in left_keys)
            changes = [
                {
                    "expected_field": left_name,
                    "observed_field": right_name,
                    "expected": left.get(left_name),
                    "observed": right.get(right_name),
                }
                for left_name, right_name in fields
                if left.get(left_name) != right.get(right_name)
            ]
            differences = [str(change["expected_field"]) for change in changes]
            status = "changed" if differences else "match"
        if left is None or right is None:
            changes = []
        item["_reconciliation_status"] = status
        item["_reconciliation_differences"] = differences
        item["_reconciliation_changes"] = changes
        item["_reconciliation_key"] = key_values
        item["_reconciliation_key_hash"] = _reconciliation_hash(
            {
                "left_keys": left_keys,
                "right_keys": right_keys,
                "values": key_values,
            }
        )
        item["_reconciliation_input_hash"] = _reconciliation_hash(
            {
                "key": key_values,
                "expected": left,
                "observed": right,
                "comparisons": comparisons,
            }
        )
        item["_reconciliation_before"] = dict(left) if left is not None else None
        item["_reconciliation_after"] = dict(right) if right is not None else None
        output.append(item)
    return output


def _reconciliation_hash(value: object) -> str:
    encoded = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        default=str,
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def _comparison_fields(value: object) -> tuple[tuple[str, str], ...]:
    if not isinstance(value, list):
        return ()
    fields: list[tuple[str, str]] = []
    for item in value:
        if isinstance(item, str) and item:
            fields.append((item, item))
        elif isinstance(item, dict):
            left = str(item.get("left") or item.get("column") or "")
            right = str(item.get("right") or left)
            if left and right:
                fields.append((left, right))
    return tuple(fields)


def _apply_reconciliation_decisions(
    records: list[dict[str, Any]],
    decisions: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> tuple[list[dict[str, Any]], tuple[str, ...]]:
    columns = {
        name: str(config[name])
        for name in (
            "decision_key_column",
            "decision_input_column",
            "decision_ref_column",
            "action_column",
            "actor_column",
            "decided_at_column",
            "reason_column",
            "correction_column",
        )
    }
    allowed_actions = {
        str(action)
        for action in config.get("allowed_actions", ())
    }
    decision_index: dict[str, dict[str, Any]] = {}
    for row_number, decision in enumerate(decisions, start=1):
        key_hash = _required_decision_hash(
            decision.get(columns["decision_key_column"]),
            row_number=row_number,
            label="key hash",
            node_id=node_id,
        )
        _required_decision_hash(
            decision.get(columns["decision_input_column"]),
            row_number=row_number,
            label="input hash",
            node_id=node_id,
        )
        for field_name, label in (
            ("decision_ref_column", "reference"),
            ("action_column", "action"),
            ("actor_column", "actor reference"),
            ("decided_at_column", "decision time"),
        ):
            _required_decision_text(
                decision.get(columns[field_name]),
                row_number=row_number,
                label=label,
                node_id=node_id,
            )
        action = str(decision[columns["action_column"]]).strip()
        if action not in allowed_actions:
            raise PipelineExecutionError(
                f"Decision row {row_number} uses an action outside the governed action set.",
                node_id=node_id,
            )
        correction = decision.get(columns["correction_column"])
        if correction is not None and not isinstance(correction, dict):
            raise PipelineExecutionError(
                f"Decision row {row_number} correction must be an object or null.",
                node_id=node_id,
            )
        if action == "correct" and not correction:
            raise PipelineExecutionError(
                f"Decision row {row_number} requires a non-empty correction object.",
                node_id=node_id,
            )
        if key_hash in decision_index:
            raise PipelineExecutionError(
                "Decision rows contain more than one current decision for a reconciliation key.",
                node_id=node_id,
            )
        decision_index[key_hash] = decision

    matched_keys: set[str] = set()
    stale_count = 0
    output: list[dict[str, Any]] = []
    for record in records:
        key_hash = str(record.get("_reconciliation_key_hash") or "")
        input_hash = str(record.get("_reconciliation_input_hash") or "")
        if not _is_sha256(key_hash) or not _is_sha256(input_hash):
            raise PipelineExecutionError(
                "Decision application requires reconciliation key and input hashes.",
                node_id=node_id,
            )
        decision = decision_index.get(key_hash)
        item = dict(record)
        if decision is None:
            _set_decision_fields(item, state="unreviewed")
            output.append(item)
            continue
        matched_keys.add(key_hash)
        decision_input_hash = str(
            decision[columns["decision_input_column"]]
        ).strip()
        state = "applied" if decision_input_hash == input_hash else "stale"
        if state == "stale":
            stale_count += 1
        _set_decision_fields(
            item,
            state=state,
            action=str(decision[columns["action_column"]]).strip(),
            decision_ref=str(
                decision[columns["decision_ref_column"]]
            ).strip(),
            actor_ref=str(decision[columns["actor_column"]]).strip(),
            decided_at=str(decision[columns["decided_at_column"]]).strip(),
            reason=_optional_decision_text(
                decision.get(columns["reason_column"])
            ),
            correction=decision.get(columns["correction_column"]),
        )
        output.append(item)

    unmatched_count = len(decision_index.keys() - matched_keys)
    messages = tuple(
        message
        for count, message in (
            (
                stale_count,
                f"{stale_count} decision(s) are stale because reconciliation inputs changed.",
            ),
            (
                unmatched_count,
                f"{unmatched_count} decision(s) no longer match a current reconciliation row.",
            ),
        )
        if count
    )
    return output, messages


def _required_decision_hash(
    value: object,
    *,
    row_number: int,
    label: str,
    node_id: str,
) -> str:
    text = str(value or "").strip()
    if not _is_sha256(text):
        raise PipelineExecutionError(
            f"Decision row {row_number} requires a SHA-256 {label}.",
            node_id=node_id,
        )
    return text


def _is_sha256(value: str) -> bool:
    if len(value) != 64:
        return False
    try:
        int(value, 16)
    except ValueError:
        return False
    return True


def _required_decision_text(
    value: object,
    *,
    row_number: int,
    label: str,
    node_id: str,
) -> str:
    text = str(value or "").strip()
    if not text:
        raise PipelineExecutionError(
            f"Decision row {row_number} requires a {label}.",
            node_id=node_id,
        )
    return text


def _optional_decision_text(value: object) -> str | None:
    text = str(value or "").strip()
    return text or None


def _set_decision_fields(
    row: dict[str, Any],
    *,
    state: str,
    action: str | None = None,
    decision_ref: str | None = None,
    actor_ref: str | None = None,
    decided_at: str | None = None,
    reason: str | None = None,
    correction: object | None = None,
) -> None:
    row["_decision_state"] = state
    row["_decision_action"] = action
    row["_decision_ref"] = decision_ref
    row["_decision_actor_ref"] = actor_ref
    row["_decision_at"] = decided_at
    row["_decision_reason"] = reason
    row["_decision_correction"] = correction


def _unique_row_index(
    rows: list[dict[str, Any]],
    columns: list[str],
    *,
    node_id: str,
    input_label: str,
) -> dict[tuple[Any, ...], dict[str, Any]]:
    index: dict[tuple[Any, ...], dict[str, Any]] = {}
    for row in rows:
        key = tuple(_hashable(row.get(column)) for column in columns)
        if key in index:
            raise PipelineExecutionError(
                f"Reconciliation keys are not unique in the {input_label} input.",
                node_id=node_id,
            )
        index[key] = row
    return index


def _execute_subflow(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    source_resolver: SourceResolver | None,
    execution_depth: int,
) -> PipelineExecutionResult:
    parameters = config.get("parameters")
    graph_payload = substitute_parameters(
        config.get("graph"),
        parameters if isinstance(parameters, dict) else {},
    )
    graph = PipelineGraph.model_validate(graph_payload)
    input_nodes = [
        node
        for node in graph.nodes
        if node.type == "source.inline" and node.config.get("input_binding") is True
    ]
    if len(input_nodes) != 1:
        raise PipelineExecutionError(
            "A reusable subflow needs exactly one inline source with input_binding=true."
        )
    input_node = input_nodes[0]
    graph = graph.model_copy(
        update={
            "nodes": [
                (
                    node.model_copy(
                        update={
                            "config": {
                                **node.config,
                                "rows": [dict(row) for row in rows],
                            }
                        },
                        deep=True,
                    )
                    if node.id == input_node.id
                    else node
                )
                for node in graph.nodes
            ]
        },
        deep=True,
    )
    return execute_preview(
        graph,
        row_limit=MAX_INTERMEDIATE_ROWS,
        source_resolver=source_resolver,
        _execution_depth=execution_depth + 1,
    )




def _join_key(row: dict[str, Any], columns: list[str]) -> tuple[Any, ...] | None:
    values = tuple(row.get(column) for column in columns)
    if any(value is None for value in values):
        return None
    try:
        hash(values)
    except TypeError:
        return tuple(
            json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
            for value in values
        )
    return values


def _ordered_columns(rows: list[dict[str, Any]]) -> list[str]:
    columns: list[str] = []
    for row in rows:
        for column in row:
            if column not in columns:
                columns.append(column)
    return columns


def _merge_join_rows(
    left_row: dict[str, Any] | None,
    right_row: dict[str, Any] | None,
    *,
    left_columns: list[str],
    right_columns: list[str],
    right_prefix: str,
) -> dict[str, Any]:
    output = {
        column: left_row.get(column) if left_row is not None else None
        for column in left_columns
    }
    for column in right_columns:
        output_column = f"{right_prefix}{column}"
        output[output_column] = right_row.get(column) if right_row is not None else None
    return output


def _guard_intermediate_size(rows: list[dict[str, Any]], *, node_id: str) -> None:
    if len(rows) > MAX_INTERMEDIATE_ROWS:
        raise PipelineExecutionError(
            f"Preview join exceeded the {MAX_INTERMEDIATE_ROWS:,}-row intermediate limit.",
            node_id=node_id,
        )


def _filter_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    column = str(config["column"])
    operator = str(config["operator"])
    expected = config.get("value")
    result: list[dict[str, Any]] = []
    for row in rows:
        actual = row.get(column)
        try:
            matches = _compare(actual, operator, expected)
        except (TypeError, ValueError) as exc:
            raise PipelineExecutionError(
                f"Cannot apply {operator!r} to column {column!r}: {exc}",
                node_id=node_id,
            ) from exc
        if matches:
            result.append(dict(row))
    return result


def _compare(actual: Any, operator: str, expected: Any) -> bool:
    if operator == "is_null":
        return actual is None
    if operator == "not_null":
        return actual is not None
    if operator == "eq":
        return actual == expected
    if operator == "ne":
        return actual != expected
    if operator == "contains":
        return expected is not None and str(expected).casefold() in str(actual or "").casefold()
    if actual is None or expected is None:
        return False
    if operator == "gt":
        return actual > expected
    if operator == "gte":
        return actual >= expected
    if operator == "lt":
        return actual < expected
    if operator == "lte":
        return actual <= expected
    raise ValueError(f"unknown operator {operator!r}")


def _select_rows(rows: list[dict[str, Any]], config: dict[str, Any]) -> list[dict[str, Any]]:
    fields = config["fields"]
    normalized = [
        (
            field if isinstance(field, str) else str(field["column"]),
            field if isinstance(field, str) else str(field.get("alias") or field["column"]),
        )
        for field in fields
    ]
    return [
        {alias: row.get(column) for column, alias in normalized}
        for row in rows
    ]


def _aggregate_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
    *,
    node_id: str,
) -> list[dict[str, Any]]:
    group_by = [str(item) for item in config.get("group_by", [])]
    aggregates = list(config["aggregates"])
    grouped = _group_rows(rows, group_by=group_by)
    return [
        _aggregate_group(
            key,
            group_rows,
            group_by=group_by,
            aggregates=aggregates,
            node_id=node_id,
        )
        for key, group_rows in grouped.items()
    ]


def _group_rows(
    rows: list[dict[str, Any]],
    *,
    group_by: list[str],
) -> dict[tuple[Any, ...], list[dict[str, Any]]]:
    grouped: dict[tuple[Any, ...], list[dict[str, Any]]] = defaultdict(list)
    if rows:
        for row in rows:
            grouped[tuple(row.get(column) for column in group_by)].append(row)
    elif not group_by:
        grouped[()] = []
    return grouped


def _aggregate_group(
    key: tuple[Any, ...],
    rows: list[dict[str, Any]],
    *,
    group_by: list[str],
    aggregates: list[dict[str, Any]],
    node_id: str,
) -> dict[str, Any]:
    result = {
        column: value
        for column, value in zip(group_by, key, strict=True)
    }
    for aggregate in aggregates:
        alias = str(aggregate["alias"])
        result[alias] = _aggregate_value(
            str(aggregate["function"]),
            aggregate.get("column"),
            rows,
            node_id=node_id,
        )
    return result


def _aggregate_value(
    function: str,
    column: Any,
    rows: list[dict[str, Any]],
    *,
    node_id: str,
) -> Any:
    values = [
        row.get(column)
        for row in rows
        if column is not None and row.get(column) is not None
    ]
    try:
        if function == "count":
            return len(rows) if column in (None, "", "*") else len(values)
        handler = _AGGREGATE_HANDLERS.get(function)
        if handler is None:
            raise ValueError(f"unknown aggregate function {function!r}")
        return handler(values)
    except (ArithmeticError, TypeError, ValueError) as exc:
        raise PipelineExecutionError(
            f"Cannot calculate {function.upper()} for {column!r}: {exc}",
            node_id=node_id,
        ) from exc


_AGGREGATE_HANDLERS: dict[str, Callable[[list[Any]], Any]] = {
    "sum": lambda values: sum(values) if values else 0,
    "avg": lambda values: sum(values) / len(values) if values else None,
    "min": lambda values: min(values) if values else None,
    "max": lambda values: max(values) if values else None,
}


def _sort_rows(rows: list[dict[str, Any]], config: dict[str, Any]) -> list[dict[str, Any]]:
    result = [dict(row) for row in rows]
    for field_config in reversed(config["fields"]):
        column = str(field_config["column"])
        reverse = field_config.get("direction", "asc") == "desc"
        concrete = [row for row in result if row.get(column) is not None]
        nulls = [row for row in result if row.get(column) is None]
        concrete.sort(key=lambda row: _sortable_value(row[column]), reverse=reverse)
        result = [*concrete, *nulls]
    return result


def _rank_rows(
    rows: list[dict[str, Any]],
    config: dict[str, Any],
) -> list[dict[str, Any]]:
    partition_by = [str(item) for item in config.get("partition_by", [])]
    order_by = list(config["order_by"])
    partitions: dict[tuple[Any, ...], list[tuple[int, dict[str, Any]]]] = (
        defaultdict(list)
    )
    for index, row in enumerate(rows):
        key = tuple(_hashable(row.get(column)) for column in partition_by)
        partitions[key].append((index, row))

    ranks: dict[int, int] = {}
    method = str(config.get("method", "row_number"))
    for partition in partitions.values():
        ordered = list(partition)
        for field_config in reversed(order_by):
            column = str(field_config["column"])
            reverse = field_config.get("direction", "asc") == "desc"
            concrete = [
                item for item in ordered if item[1].get(column) is not None
            ]
            nulls = [
                item for item in ordered if item[1].get(column) is None
            ]
            concrete.sort(
                key=lambda item: _sortable_value(item[1][column]),
                reverse=reverse,
            )
            ordered = [*concrete, *nulls]

        previous_values: tuple[Any, ...] | None = None
        current_rank = 0
        dense_rank = 0
        for position, (source_index, row) in enumerate(ordered, start=1):
            values = tuple(
                _hashable(row.get(str(field["column"])))
                for field in order_by
            )
            if previous_values is None or values != previous_values:
                current_rank = position
                dense_rank += 1
                previous_values = values
            ranks[source_index] = (
                position
                if method == "row_number"
                else dense_rank
                if method == "dense_rank"
                else current_rank
            )

    target = str(config["target_column"])
    return [
        {**row, target: ranks[index]}
        for index, row in enumerate(rows)
    ]


def _sortable_value(value: Any) -> tuple[str, Any]:
    if isinstance(value, (int, float, Decimal, str)):
        return type(value).__name__, value
    return type(value).__name__, str(value)


def _hashable(value: Any) -> Any:
    if isinstance(value, (dict, list, tuple, set)):
        return json.dumps(value, sort_keys=True, default=str)
    return value


def _rows_fingerprint(rows: list[dict[str, Any]]) -> str:
    encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def _type_name(value: Any) -> str:
    if value is None:
        return "unknown"
    if isinstance(value, bool):
        return "boolean"
    if isinstance(value, int):
        return "integer"
    if isinstance(value, (float, Decimal)):
        return "number"
    if isinstance(value, str):
        return "string"
    if isinstance(value, list):
        return "array"
    if isinstance(value, dict):
        return "object"
    return type(value).__name__.lower()


def _execute_inline_source(
    context: OperatorExecutionContext,
) -> OperatorExecutionResult:
    rows = [dict(row) for row in context.node.config.get("rows", [])]
    return OperatorExecutionResult(
        rows=rows,
        input_row_count=len(rows),
        source_fingerprints=(
            {
                "node_id": context.node.id,
                "source_name": context.node.config.get("source_name"),
                "kind": "inline",
                "fingerprint": _rows_fingerprint(rows),
                "row_count": len(rows),
            },
        ),
    )


def _execute_reference_source(
    context: OperatorExecutionContext,
) -> OperatorExecutionResult:
    if context.source_resolver is None:
        raise PipelineExecutionError(
            "Datasource-backed preview requires the Datasources catalogue capability.",
            node_id=context.node.id,
        )
    resolved = context.source_resolver(context.node, MAX_SOURCE_ROWS)
    rows = [dict(row) for row in resolved.rows]
    messages = (
        (
            f"Source preview used {len(rows):,} of "
            f"{resolved.total_rows:,} rows."
        ),
    ) if resolved.truncated else ()
    return OperatorExecutionResult(
        rows=rows,
        messages=messages,
        input_row_count=len(rows),
        source_fingerprints=(
            {
                "node_id": context.node.id,
                "source_ref": resolved.source_ref,
                "source_name": context.node.config.get("source_name"),
                "kind": "datasource",
                "provider": resolved.provider,
                "fingerprint": resolved.fingerprint,
                "row_count": resolved.total_rows,
                "preview_rows": len(rows),
                "truncated": resolved.truncated,
            },
        ),
    )


def _execute_subflow_node(
    context: OperatorExecutionContext,
) -> OperatorExecutionResult:
    nested = _execute_subflow(
        context.input_rows,
        context.node.config,
        source_resolver=context.source_resolver,
        execution_depth=context.execution_depth,
    )
    return OperatorExecutionResult(
        rows=nested.rows,
        messages=tuple(
            f"Subflow {item.node_id}: {message}"
            for item in nested.node_diagnostics
            for message in item.messages
        ),
        source_fingerprints=tuple(
            {
                **fingerprint,
                "subflow_node_id": context.node.id,
            }
            for fingerprint in nested.source_fingerprints
        ),
    )


def _execute_reconciliation_decisions(
    context: OperatorExecutionContext,
) -> OperatorExecutionResult:
    rows, messages = _apply_reconciliation_decisions(
        context.outputs[context.inputs_by_port["records"][0]],
        context.outputs[context.inputs_by_port["decisions"][0]],
        context.node.config,
        node_id=context.node.id,
    )
    return OperatorExecutionResult(rows=rows, messages=messages)


def _register_executors() -> None:
    executors = {
        "source.inline": _execute_inline_source,
        "source.reference": _execute_reference_source,
        "combine.union": lambda context: OperatorExecutionResult(
            rows=_union_rows(
                [
                    context.outputs[source_id]
                    for source_id in context.inputs_by_port.get("input", ())
                ],
                context.node.config,
            )
        ),
        "combine.join": lambda context: OperatorExecutionResult(
            rows=_join_rows(
                context.outputs[context.inputs_by_port["left"][0]],
                context.outputs[context.inputs_by_port["right"][0]],
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "filter": lambda context: OperatorExecutionResult(
            rows=_filter_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "filter.expression": lambda context: OperatorExecutionResult(
            rows=_expression_filter_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "distinct": lambda context: OperatorExecutionResult(
            rows=_distinct_rows(context.input_rows, context.node.config)
        ),
        "select": lambda context: OperatorExecutionResult(
            rows=_select_rows(context.input_rows, context.node.config)
        ),
        "derive": lambda context: OperatorExecutionResult(
            rows=_derive_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "expression": lambda context: OperatorExecutionResult(
            rows=_expression_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "calculate": lambda context: OperatorExecutionResult(
            rows=_calculation_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "convert": lambda context: OperatorExecutionResult(
            rows=_convert_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "replace": lambda context: OperatorExecutionResult(
            rows=_replace_rows(context.input_rows, context.node.config)
        ),
        "aggregate": lambda context: OperatorExecutionResult(
            rows=_aggregate_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "sort": lambda context: OperatorExecutionResult(
            rows=_sort_rows(context.input_rows, context.node.config)
        ),
        "window.rank": lambda context: OperatorExecutionResult(
            rows=_rank_rows(context.input_rows, context.node.config)
        ),
        "limit": lambda context: OperatorExecutionResult(
            rows=context.input_rows[: int(context.node.config["count"])]
        ),
        "quality.rules": lambda context: OperatorExecutionResult(
            rows=_quality_rows(
                context.input_rows,
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "reconcile.compare": lambda context: OperatorExecutionResult(
            rows=_reconcile_rows(
                context.outputs[context.inputs_by_port["left"][0]],
                context.outputs[context.inputs_by_port["right"][0]],
                context.node.config,
                node_id=context.node.id,
            )
        ),
        "reconcile.decisions": _execute_reconciliation_decisions,
        "subflow": _execute_subflow_node,
        "output": lambda context: OperatorExecutionResult(
            rows=[dict(row) for row in context.input_rows]
        ),
    }
    for node_type, executor in executors.items():
        if OPERATOR_REGISTRY.executor(node_type) is None:
            OPERATOR_REGISTRY.register_executor(node_type, executor)


_register_executors()


__all__ = [
    "EXECUTOR_VERSION",
    "MAX_SOURCE_ROWS",
    "PipelineExecutionError",
    "PipelineExecutionResult",
    "ResolvedSource",
    "SourceResolver",
    "execute_preview",
    "infer_columns",
]