from __future__ import annotations import hashlib import json import re from collections import deque from dataclasses import dataclass from typing import Any from govoplan_core.core.definition_graphs import ( DefinitionEdge, DefinitionNode, validate_definition_graph, ) from govoplan_dataflow.backend.node_library import ( DATAFLOW_GRAPH_LIBRARY, NODE_TYPES, node_definition, ) from govoplan_dataflow.backend.schemas import DataflowDiagnostic, GraphEdge, GraphNode, PipelineGraph SUPPORTED_NODE_TYPES = frozenset(NODE_TYPES) FILTER_OPERATORS = frozenset( {"eq", "ne", "gt", "gte", "lt", "lte", "contains", "is_null", "not_null"} ) AGGREGATE_FUNCTIONS = frozenset({"count", "sum", "avg", "min", "max"}) DERIVE_OPERATIONS = frozenset( { "copy", "upper", "lower", "trim", "concat", "coalesce", "add", "subtract", "multiply", "divide", } ) JOIN_TYPES = frozenset({"inner", "left", "right", "full"}) def canonical_graph_payload(graph: PipelineGraph) -> dict[str, Any]: return graph.model_dump(mode="json", exclude_none=True) def definition_hash(graph: PipelineGraph, sql_text: str | None = None) -> str: payload = { "graph": canonical_graph_payload(graph), "sql_text": (sql_text or "").strip() or None, } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() def preserve_compatible_graph_layout( reference: PipelineGraph, compiled: PipelineGraph, ) -> PipelineGraph: """Keep graph identity and layout when SQL preserves the same topology.""" reference_keys = _structural_node_keys(reference) compiled_keys = _structural_node_keys(compiled) if reference_keys is None or compiled_keys is None: return compiled if ( len(set(reference_keys.values())) != len(reference_keys) or len(set(compiled_keys.values())) != len(compiled_keys) or set(reference_keys.values()) != set(compiled_keys.values()) ): return compiled compiled_by_key = { key: next(node for node in compiled.nodes if node.id == node_id) for node_id, key in compiled_keys.items() } reference_edges = { _structural_edge_key(edge, reference_keys): edge for edge in reference.edges } compiled_edges = { _structural_edge_key(edge, compiled_keys): edge for edge in compiled.edges } if ( len(reference_edges) != len(reference.edges) or len(compiled_edges) != len(compiled.edges) or set(reference_edges) != set(compiled_edges) ): return compiled nodes = [] for reference_node in reference.nodes: node = compiled_by_key[reference_keys[reference_node.id]] nodes.append( node.model_copy( update={ "id": reference_node.id, "label": reference_node.label, "position": reference_node.position.model_copy(deep=True), }, deep=True, ) ) edges = [] for reference_edge in reference.edges: edge_key = _structural_edge_key(reference_edge, reference_keys) edge = compiled_edges[edge_key] edges.append( edge.model_copy( update={ "id": reference_edge.id, "source": reference_edge.source, "target": reference_edge.target, }, deep=True, ) ) return compiled.model_copy( update={"nodes": nodes, "edges": edges}, deep=True, ) def validate_graph(graph: PipelineGraph) -> list[DataflowDiagnostic]: diagnostics = [ DataflowDiagnostic( severity=item.severity, code=item.code, message=item.message, node_id=item.node_id, field=item.field, ) for item in validate_definition_graph( DATAFLOW_GRAPH_LIBRARY, nodes=tuple(DefinitionNode(id=node.id, type=node.type) for node in graph.nodes), edges=tuple( DefinitionEdge( id=edge.id, source=edge.source, target=edge.target, source_port=edge.source_port, target_port=edge.target_port, ) for edge in graph.edges ), ) ] nodes = {node.id: node for node in graph.nodes} incoming: dict[str, list[GraphEdge]] = {node_id: [] for node_id in nodes} outgoing: dict[str, list[str]] = {node_id: [] for node_id in nodes} for edge in graph.edges: if edge.source not in nodes or edge.target not in nodes or edge.source == edge.target: continue source_definition = node_definition(nodes[edge.source].type) target_definition = node_definition(nodes[edge.target].type) if ( source_definition is None or target_definition is None or edge.source_port not in {port.id for port in source_definition.output_ports} or edge.target_port not in {port.id for port in target_definition.input_ports} ): continue outgoing[edge.source].append(edge.target) incoming[edge.target].append(edge) source_nodes = [node for node in graph.nodes if node.type.startswith("source.")] output_nodes = [node for node in graph.nodes if node.type == "output"] source_names = [ str(node.config.get("source_name", "")).strip() for node in source_nodes if _non_empty_text(node.config.get("source_name")) ] duplicate_source_names = sorted( { name for name in source_names if sum(candidate.casefold() == name.casefold() for candidate in source_names) > 1 }, key=str.casefold, ) if duplicate_source_names: diagnostics.append( _error( "source.duplicate_name", "Logical source names must be unique: " f"{', '.join(duplicate_source_names)}.", ) ) for node in graph.nodes: if node.type not in SUPPORTED_NODE_TYPES: continue diagnostics.extend(_validate_node_config(node)) ordered, cyclic = topological_order(graph) if not cyclic and source_nodes and len(output_nodes) == 1: reachable: set[str] = set() for source in source_nodes: reachable.update(_reachable_from(source.id, outgoing)) if len(reachable) != len(nodes): diagnostics.append( _error("graph.disconnected", "Every node must be connected to a pipeline source.") ) reverse_adjacency = { node_id: [edge.source for edge in edges] for node_id, edges in incoming.items() } reaches_output = _reachable_from(output_nodes[0].id, reverse_adjacency) if len(reaches_output) != len(nodes): diagnostics.append( _error("graph.dead_end", "Every node must lead to the pipeline output.") ) if ordered and ordered[-1] != output_nodes[0].id: diagnostics.append( _error("graph.output_not_terminal", "The output node must be the terminal transform.") ) diagnostics.extend(_validate_graph_schemas(graph, ordered=ordered)) return _dedupe_diagnostics(diagnostics) def _structural_node_keys( graph: PipelineGraph, ) -> dict[str, tuple[object, ...]] | None: node_by_id = {node.id: node for node in graph.nodes} if ( len(node_by_id) != len(graph.nodes) or any( edge.source not in node_by_id or edge.target not in node_by_id for edge in graph.edges ) ): return None ordered, cyclic = topological_order(graph) if cyclic: return None incoming: dict[str, list[GraphEdge]] = { node.id: [] for node in graph.nodes } for edge in graph.edges: if edge.target in incoming: incoming[edge.target].append(edge) keys: dict[str, tuple[object, ...]] = {} for node_id in ordered: node = node_by_id[node_id] input_keys: list[tuple[object, ...]] = [] for edge in incoming[node_id]: source_key = keys.get(edge.source) if source_key is None: return None input_keys.append( ( edge.target_port, edge.source_port, source_key, ) ) if node.type != "combine.union": input_keys.sort(key=repr) source_name = ( str(node.config.get("source_name", "")).strip().casefold() if node.type.startswith("source.") else None ) keys[node_id] = ( node.type, source_name, tuple(input_keys), ) return keys def _structural_edge_key( edge: GraphEdge, node_keys: dict[str, tuple[object, ...]], ) -> tuple[object, ...]: return ( node_keys[edge.source], node_keys[edge.target], edge.source_port, edge.target_port, ) def _dedupe_diagnostics( diagnostics: list[DataflowDiagnostic], ) -> list[DataflowDiagnostic]: seen: set[tuple[str, str | None, str | None]] = set() result: list[DataflowDiagnostic] = [] for item in diagnostics: key = (item.code, item.node_id, item.field) if key in seen: continue seen.add(key) result.append(item) return result def topological_order(graph: PipelineGraph) -> tuple[list[str], bool]: node_ids = [node.id for node in graph.nodes] incoming_count = {node_id: 0 for node_id in node_ids} outgoing: dict[str, list[str]] = {node_id: [] for node_id in node_ids} for edge in graph.edges: if edge.source in outgoing and edge.target in incoming_count and edge.source != edge.target: outgoing[edge.source].append(edge.target) incoming_count[edge.target] += 1 ready = deque(node_id for node_id in node_ids if incoming_count[node_id] == 0) ordered: list[str] = [] while ready: node_id = ready.popleft() ordered.append(node_id) for target in outgoing[node_id]: incoming_count[target] -= 1 if incoming_count[target] == 0: ready.append(target) return ordered, len(ordered) != len(node_ids) def graph_inputs_by_port(graph: PipelineGraph) -> dict[str, dict[str, list[str]]]: result: dict[str, dict[str, list[str]]] = {} for edge in graph.edges: result.setdefault(edge.target, {}).setdefault(edge.target_port, []).append(edge.source) return result def _reachable_from(start: str, adjacency: dict[str, list[str]]) -> set[str]: seen: set[str] = set() pending = [start] while pending: node_id = pending.pop() if node_id in seen: continue seen.add(node_id) pending.extend(adjacency.get(node_id, ())) return seen @dataclass(frozen=True) class _SchemaState: columns: frozenset[str] open: bool = False def knows(self, column: str) -> bool: return self.open or column in self.columns def _validate_graph_schemas( graph: PipelineGraph, *, ordered: list[str], ) -> list[DataflowDiagnostic]: diagnostics: list[DataflowDiagnostic] = [] node_by_id = {node.id: node for node in graph.nodes} inputs = graph_inputs_by_port(graph) schemas: dict[str, _SchemaState] = {} for node_id in ordered: node = node_by_id[node_id] node_inputs = inputs.get(node.id, {}) input_states = [ schemas[source_id] for port_sources in node_inputs.values() for source_id in port_sources if source_id in schemas ] input_state = input_states[0] if input_states else _SchemaState(frozenset(), open=True) if node.type == "source.inline": rows = node.config.get("rows") columns = { str(column) for row in rows if isinstance(rows, list) and isinstance(row, dict) for column in row } if isinstance(rows, list) else set() schemas[node.id] = _SchemaState( frozenset(columns), open=not columns, ) continue if node.type == "source.reference": columns = _configured_source_columns(node.config.get("source_columns")) schemas[node.id] = _SchemaState( frozenset(columns), open=not columns, ) continue if node.type == "combine.union": if len(input_states) > 1: closed_shapes = { state.columns for state in input_states if not state.open } if len(closed_shapes) > 1: diagnostics.append( _warning( "union.schema_mismatch", "Appended inputs use different columns; missing values will be null.", node_id=node.id, ) ) schemas[node.id] = _SchemaState( frozenset().union(*(state.columns for state in input_states)), open=any(state.open for state in input_states), ) continue if node.type == "combine.join": left_state = _port_schema(node_inputs, schemas, "left") right_state = _port_schema(node_inputs, schemas, "right") _validate_columns( diagnostics, node=node, state=left_state, columns=node.config.get("left_keys"), field="left_keys", ) _validate_columns( diagnostics, node=node, state=right_state, columns=node.config.get("right_keys"), field="right_keys", ) prefix = str(node.config.get("right_prefix", "right_")) prefixed_right = {f"{prefix}{column}" for column in right_state.columns} collisions = left_state.columns & prefixed_right if collisions: diagnostics.append( _error( "join.output_collision", f"Join output columns collide: {', '.join(sorted(collisions))}.", node_id=node.id, field="right_prefix", ) ) schemas[node.id] = _SchemaState( frozenset(left_state.columns | prefixed_right), open=left_state.open or right_state.open, ) continue if node.type == "filter": _validate_columns( diagnostics, node=node, state=input_state, columns=[node.config.get("column")], field="column", ) schemas[node.id] = input_state continue if node.type == "distinct": _validate_columns( diagnostics, node=node, state=input_state, columns=node.config.get("columns"), field="columns", ) schemas[node.id] = input_state continue if node.type == "select": fields = node.config.get("fields") selected_columns: list[str] = [] output_columns: list[str] = [] if isinstance(fields, list): for field in fields: if isinstance(field, str): selected_columns.append(field) output_columns.append(field) elif isinstance(field, dict): column = field.get("column") alias = field.get("alias") or column if isinstance(column, str): selected_columns.append(column) if isinstance(alias, str): output_columns.append(alias) _validate_columns( diagnostics, node=node, state=input_state, columns=selected_columns, field="fields", ) _validate_output_names( diagnostics, node=node, columns=output_columns, field="fields", ) schemas[node.id] = _SchemaState(frozenset(output_columns)) continue if node.type == "derive": _validate_columns( diagnostics, node=node, state=input_state, columns=node.config.get("source_columns"), field="source_columns", ) target = node.config.get("target_column") if isinstance(target, str) and target: if target in input_state.columns: diagnostics.append( _warning( "derive.overwrites_column", f"Derived column {target!r} replaces an existing value.", node_id=node.id, field="target_column", ) ) schemas[node.id] = _SchemaState( input_state.columns | frozenset((target,)), open=input_state.open, ) else: schemas[node.id] = input_state continue if node.type == "aggregate": group_by = _text_items(node.config.get("group_by")) aggregates = node.config.get("aggregates") aggregate_columns = [ str(item.get("column")) for item in aggregates if isinstance(aggregates, list) and isinstance(item, dict) and item.get("column") not in (None, "", "*") ] if isinstance(aggregates, list) else [] _validate_columns( diagnostics, node=node, state=input_state, columns=[*group_by, *aggregate_columns], field="aggregates", ) aliases = [ str(item.get("alias")) for item in aggregates if isinstance(aggregates, list) and isinstance(item, dict) and item.get("alias") ] if isinstance(aggregates, list) else [] output_columns = [*group_by, *aliases] _validate_output_names( diagnostics, node=node, columns=output_columns, field="aggregates", ) schemas[node.id] = _SchemaState(frozenset(output_columns)) continue if node.type == "sort": fields = node.config.get("fields") columns = [ str(item.get("column")) for item in fields if isinstance(fields, list) and isinstance(item, dict) and item.get("column") ] if isinstance(fields, list) else [] _validate_columns( diagnostics, node=node, state=input_state, columns=columns, field="fields", ) schemas[node.id] = input_state continue schemas[node.id] = input_state return diagnostics def _configured_source_columns(value: object) -> set[str]: if not isinstance(value, list): return set() return { item if isinstance(item, str) else str(item.get("name")) for item in value if ( isinstance(item, str) and item or isinstance(item, dict) and item.get("name") ) } def _port_schema( inputs: dict[str, list[str]], schemas: dict[str, _SchemaState], port: str, ) -> _SchemaState: source_ids = inputs.get(port, []) return schemas.get(source_ids[0], _SchemaState(frozenset(), open=True)) if source_ids else _SchemaState(frozenset(), open=True) def _validate_columns( diagnostics: list[DataflowDiagnostic], *, node: GraphNode, state: _SchemaState, columns: object, field: str, ) -> None: for column in _text_items(columns): if not state.knows(column): diagnostics.append( _error( "schema.unknown_column", f"Column {column!r} is not available at this node.", node_id=node.id, field=field, ) ) def _validate_output_names( diagnostics: list[DataflowDiagnostic], *, node: GraphNode, columns: list[str], field: str, ) -> None: duplicates = sorted({column for column in columns if columns.count(column) > 1}) if duplicates: diagnostics.append( _error( "schema.duplicate_output", f"Output column names must be unique: {', '.join(duplicates)}.", node_id=node.id, field=field, ) ) def _text_items(value: object) -> list[str]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, str) and item] def _validate_node_config(node: GraphNode) -> list[DataflowDiagnostic]: config = node.config diagnostics: list[DataflowDiagnostic] = [] if node.type in {"source.inline", "source.reference"}: source_name = config.get("source_name") if not isinstance(source_name, str) or not source_name.strip(): diagnostics.append( _error( "source.name_required", "A source needs a logical SQL name.", node_id=node.id, field="source_name", ) ) elif re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", source_name.strip()) is None: diagnostics.append( _error( "source.name_invalid", "Logical source names must be SQL identifiers, such as monthly_cases.", node_id=node.id, field="source_name", ) ) if node.type == "source.reference": if not _non_empty_text(config.get("source_ref")): diagnostics.append( _error( "source.reference_required", "Choose a datasource.", node_id=node.id, field="source_ref", ) ) expected_fingerprint = config.get("expected_fingerprint") if expected_fingerprint is not None and not isinstance(expected_fingerprint, str): diagnostics.append( _error( "source.fingerprint", "The expected source fingerprint must be text.", node_id=node.id, field="expected_fingerprint", ) ) return diagnostics rows = config.get("rows") if not isinstance(rows, list): diagnostics.append( _error("source.rows_required", "Inline source rows must be a list.", node_id=node.id, field="rows") ) elif len(rows) > 250: diagnostics.append( _error("source.row_limit", "Inline sources are limited to 250 rows.", node_id=node.id, field="rows") ) elif any(not isinstance(row, dict) for row in rows): diagnostics.append( _error("source.row_shape", "Every inline source row must be an object.", node_id=node.id, field="rows") ) if len(json.dumps(config, default=str).encode("utf-8")) > 512_000: diagnostics.append( _error("source.byte_limit", "Inline source configuration is limited to 512 KiB.", node_id=node.id) ) elif node.type == "filter": if not _non_empty_text(config.get("column")): diagnostics.append(_node_field_error(node, "filter.column", "Choose a column.", "column")) operator = config.get("operator") if operator not in FILTER_OPERATORS: diagnostics.append( _node_field_error(node, "filter.operator", "Choose a supported comparison operator.", "operator") ) if operator not in {"is_null", "not_null"} and "value" not in config: diagnostics.append(_node_field_error(node, "filter.value", "Enter a comparison value.", "value")) elif node.type == "distinct": columns = config.get("columns", []) if not isinstance(columns, list) or any(not _non_empty_text(item) for item in columns): diagnostics.append( _node_field_error( node, "distinct.columns", "Distinct key columns must be named.", "columns", ) ) elif node.type == "combine.union": if config.get("mode", "all") not in {"all", "distinct"}: diagnostics.append( _node_field_error( node, "union.mode", "Choose whether duplicate rows are kept or removed.", "mode", ) ) elif node.type == "combine.join": if config.get("join_type", "inner") not in JOIN_TYPES: diagnostics.append( _node_field_error(node, "join.type", "Choose a supported join type.", "join_type") ) left_keys = config.get("left_keys") right_keys = config.get("right_keys") if ( not isinstance(left_keys, list) or not left_keys or any(not _non_empty_text(item) for item in left_keys) ): diagnostics.append( _node_field_error(node, "join.left_keys", "Add at least one left key.", "left_keys") ) if ( not isinstance(right_keys, list) or not right_keys or any(not _non_empty_text(item) for item in right_keys) ): diagnostics.append( _node_field_error(node, "join.right_keys", "Add at least one right key.", "right_keys") ) if isinstance(left_keys, list) and isinstance(right_keys, list) and len(left_keys) != len(right_keys): diagnostics.append( _node_field_error( node, "join.key_count", "Left and right joins need the same number of keys.", "right_keys", ) ) right_prefix = config.get("right_prefix", "right_") if ( not _non_empty_text(right_prefix) or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*_", str(right_prefix)) is None ): diagnostics.append( _node_field_error( node, "join.right_prefix", "Enter an identifier prefix ending in an underscore, such as right_.", "right_prefix", ) ) elif node.type == "select": fields = config.get("fields") if not isinstance(fields, list) or not fields: diagnostics.append( _node_field_error(node, "select.fields", "Select at least one output column.", "fields") ) else: for field in fields: column = field if isinstance(field, str) else field.get("column") if isinstance(field, dict) else None if not _non_empty_text(column): diagnostics.append( _node_field_error(node, "select.field", "Selected fields need a source column.", "fields") ) break elif node.type == "aggregate": group_by = config.get("group_by", []) aggregates = config.get("aggregates") if not isinstance(group_by, list) or any(not _non_empty_text(item) for item in group_by): diagnostics.append( _node_field_error(node, "aggregate.group_by", "Group-by columns must be named.", "group_by") ) if not isinstance(aggregates, list) or not aggregates: diagnostics.append( _node_field_error(node, "aggregate.required", "Add at least one aggregate.", "aggregates") ) else: for aggregate in aggregates: if not isinstance(aggregate, dict) or aggregate.get("function") not in AGGREGATE_FUNCTIONS: diagnostics.append( _node_field_error( node, "aggregate.function", "Choose COUNT, SUM, AVG, MIN, or MAX.", "aggregates", ) ) break if aggregate.get("function") != "count" and not _non_empty_text(aggregate.get("column")): diagnostics.append( _node_field_error( node, "aggregate.column", "This aggregate needs a source column.", "aggregates", ) ) break if not _non_empty_text(aggregate.get("alias")): diagnostics.append( _node_field_error(node, "aggregate.alias", "Every aggregate needs an alias.", "aggregates") ) break elif node.type == "derive": if not _non_empty_text(config.get("target_column")): diagnostics.append( _node_field_error( node, "derive.target_column", "Choose an output column.", "target_column", ) ) operation = config.get("operation") if operation not in DERIVE_OPERATIONS: diagnostics.append( _node_field_error( node, "derive.operation", "Choose a supported derive operation.", "operation", ) ) source_columns = config.get("source_columns") if ( not isinstance(source_columns, list) or not source_columns or any(not _non_empty_text(item) for item in source_columns) ): diagnostics.append( _node_field_error( node, "derive.source_columns", "Choose at least one source column.", "source_columns", ) ) if operation in {"copy", "upper", "lower", "trim"} and isinstance(source_columns, list) and len(source_columns) != 1: diagnostics.append( _node_field_error( node, "derive.source_count", "This operation requires exactly one source column.", "source_columns", ) ) if operation in {"add", "subtract", "multiply", "divide"} and isinstance(source_columns, list) and len(source_columns) != 2: diagnostics.append( _node_field_error( node, "derive.numeric_source_count", "Numeric operations require exactly two source columns.", "source_columns", ) ) elif node.type == "sort": fields = config.get("fields") if not isinstance(fields, list) or not fields: diagnostics.append(_node_field_error(node, "sort.fields", "Add at least one sort field.", "fields")) elif any( not isinstance(item, dict) or not _non_empty_text(item.get("column")) or item.get("direction", "asc") not in {"asc", "desc"} for item in fields ): diagnostics.append( _node_field_error(node, "sort.field", "Sort fields need a column and direction.", "fields") ) elif node.type == "limit": count = config.get("count") if not isinstance(count, int) or isinstance(count, bool) or not 1 <= count <= 100_000: diagnostics.append( _node_field_error(node, "limit.count", "Limit must be between 1 and 100,000.", "count") ) return diagnostics def _non_empty_text(value: object) -> bool: return isinstance(value, str) and bool(value.strip()) def _node_field_error(node: GraphNode, code: str, message: str, field: str) -> DataflowDiagnostic: return _error(code, message, node_id=node.id, field=field) def _error( code: str, message: str, *, node_id: str | None = None, field: str | None = None, ) -> DataflowDiagnostic: return DataflowDiagnostic( severity="error", code=code, message=message, node_id=node_id, field=field, ) def _warning( code: str, message: str, *, node_id: str | None = None, field: str | None = None, ) -> DataflowDiagnostic: return DataflowDiagnostic( severity="warning", code=code, message=message, node_id=node_id, field=field, ) __all__ = [ "AGGREGATE_FUNCTIONS", "DERIVE_OPERATIONS", "FILTER_OPERATORS", "JOIN_TYPES", "SUPPORTED_NODE_TYPES", "canonical_graph_payload", "definition_hash", "graph_inputs_by_port", "topological_order", "validate_graph", ]