From e54225a86e47b4b9772186448102a9fad5b3ecae Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 18:36:09 +0200 Subject: [PATCH] Preserve dataflow layout across SQL edits --- README.md | 3 + src/govoplan_dataflow/backend/graph.py | 132 ++++++++++++++++++ src/govoplan_dataflow/backend/service.py | 9 +- src/govoplan_dataflow/backend/sql_compiler.py | 74 ++++++++-- tests/test_graph_and_sql.py | 114 +++++++++++++++ .../scripts/test-dataflow-page-structure.mjs | 8 ++ webui/src/features/dataflow/DataflowPage.tsx | 7 +- webui/src/styles/dataflow.css | 3 +- 8 files changed, 331 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 6e2be8c..5c37444 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,9 @@ Join nodes have explicit left and right ports. Append nodes accept two or more inputs. Derived columns use a constrained operation catalogue rather than arbitrary code. The graph editor validates manual, proximity-created, and reconnected edges against the same port, multiplicity, and cycle rules. +SQL round-trips retain node and edge identity, labels, and coordinates while +the graph topology remains compatible. Structural changes use a compact +layered layout with bounded branch and edge spacing. Governed inputs are resolved through the versioned Core capability `datasources.catalogue`; Dataflow imports neither Datasources nor Connectors and diff --git a/src/govoplan_dataflow/backend/graph.py b/src/govoplan_dataflow/backend/graph.py index e15ef8b..0d59f86 100644 --- a/src/govoplan_dataflow/backend/graph.py +++ b/src/govoplan_dataflow/backend/graph.py @@ -55,6 +55,74 @@ def definition_hash(graph: PipelineGraph, sql_text: str | None = None) -> str: 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( @@ -151,6 +219,70 @@ def validate_graph(graph: PipelineGraph) -> list[DataflowDiagnostic]: 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]: diff --git a/src/govoplan_dataflow/backend/service.py b/src/govoplan_dataflow/backend/service.py index a19ddab..884ff1f 100644 --- a/src/govoplan_dataflow/backend/service.py +++ b/src/govoplan_dataflow/backend/service.py @@ -40,7 +40,12 @@ from govoplan_dataflow.backend.governance import ( definition_governance_payload, require_definition_action, ) -from govoplan_dataflow.backend.graph import canonical_graph_payload, definition_hash, validate_graph +from govoplan_dataflow.backend.graph import ( + canonical_graph_payload, + definition_hash, + preserve_compatible_graph_layout, + validate_graph, +) from govoplan_dataflow.backend.schemas import ( DataflowDiagnostic, GraphNode, @@ -488,6 +493,8 @@ def compile_sql_draft(payload: PipelineDraftRequest) -> PipelineSqlResponse: payload.sql_text, source_nodes=_source_nodes(payload.graph, payload.source_nodes), ) + if payload.graph is not None: + graph = preserve_compatible_graph_layout(payload.graph, graph) except SqlCompilationError as exc: return PipelineSqlResponse( valid=False, diff --git a/src/govoplan_dataflow/backend/sql_compiler.py b/src/govoplan_dataflow/backend/sql_compiler.py index 259fc77..60f2b63 100644 --- a/src/govoplan_dataflow/backend/sql_compiler.py +++ b/src/govoplan_dataflow/backend/sql_compiler.py @@ -22,6 +22,12 @@ class SqlCompilationError(ValueError): self.diagnostics = diagnostics +LAYOUT_ORIGIN_X = 60 +LAYOUT_CENTER_Y = 180 +LAYOUT_LAYER_GAP = 240 +LAYOUT_BRANCH_GAP = 120 + + def compile_sql( sql_text: str, *, @@ -90,12 +96,13 @@ def compile_sql( [_sql_error("sql.union_join", "JOIN outside UNION BY NAME is not supported.")] ) union_tables, union_mode = _union_tables(union_expression) + branch_positions = _branch_positions(len(union_tables)) for index, table in enumerate(union_tables, start=1): source = _source_node( table, source_node_list, fallback_id=f"source-{index}", - position=GraphPosition(x=60, y=80 + ((index - 1) * 140)), + position=branch_positions[index - 1], ) if any(existing.id == source.id for existing in nodes): raise SqlCompilationError( @@ -111,7 +118,11 @@ def compile_sql( id="union", type="combine.union", label="Append rows", - position=GraphPosition(x=300, y=80 + ((len(nodes) - 1) * 70)), + position=_position( + 1, + y=sum(position.y for position in branch_positions) + / len(branch_positions), + ), config={"mode": union_mode}, ) nodes.append(union_node) @@ -125,6 +136,7 @@ def compile_sql( if source.type.startswith("source.") ) previous_node_id = union_node.id + next_transform_layer = 2 else: left_table = from_clause.this if not isinstance(left_table, exp.Table): @@ -146,16 +158,22 @@ def compile_sql( left_table, source_node_list, fallback_id="source-left" if right_table is not None else "source", - position=GraphPosition(x=60, y=120 if right_table is not None else 180), + position=( + _branch_positions(2)[0] + if right_table is not None + else _position(0) + ), ) nodes.append(left_source) previous_node_id = left_source.id + next_transform_layer = 1 if isinstance(right_table, exp.Table): + join_positions = _branch_positions(2) right_source = _source_node( right_table, source_node_list, fallback_id="source-right", - position=GraphPosition(x=60, y=280), + position=join_positions[1], ) if right_source.id == left_source.id: raise SqlCompilationError( @@ -168,7 +186,7 @@ def compile_sql( id="join", type="combine.join", label=f"Join {left_table.name} and {right_table.name}", - position=GraphPosition(x=300, y=200), + position=_position(1), config=join_config, ) nodes.extend((right_source, join_node)) @@ -194,6 +212,13 @@ def compile_sql( right_table, right_prefix=right_prefix, ) + next_transform_layer = 2 + + def next_transform_position() -> GraphPosition: + nonlocal next_transform_layer + position = _position(next_transform_layer) + next_transform_layer += 1 + return position def append_transform(node: GraphNode) -> None: nonlocal previous_node_id @@ -214,7 +239,7 @@ def compile_sql( id=f"filter-{index}", type="filter", label=f"Filter {index}", - position=_position(len(nodes)), + position=next_transform_position(), config=_condition_config( condition, qualifier_prefixes=qualifier_prefixes, @@ -286,7 +311,7 @@ def compile_sql( id="aggregate", type="aggregate", label="Aggregate", - position=_position(len(nodes)), + position=next_transform_position(), config={"group_by": group_by, "aggregates": aggregate_specs}, ) ) @@ -296,7 +321,7 @@ def compile_sql( id="select", type="select", label="Select columns", - position=_position(len(nodes)), + position=next_transform_position(), config={"fields": projection_fields}, ) ) @@ -311,7 +336,7 @@ def compile_sql( id="distinct", type="distinct", label="Remove duplicates", - position=_position(len(nodes)), + position=next_transform_position(), config={"columns": []}, ) ) @@ -337,7 +362,7 @@ def compile_sql( id="sort", type="sort", label="Sort", - position=_position(len(nodes)), + position=next_transform_position(), config={"fields": fields}, ) ) @@ -360,7 +385,7 @@ def compile_sql( id="limit", type="limit", label="Limit", - position=_position(len(nodes)), + position=next_transform_position(), config={"count": count}, ) ) @@ -370,7 +395,7 @@ def compile_sql( id="output", type="output", label="Preview output", - position=_position(len(nodes)), + position=next_transform_position(), config={}, ) ) @@ -801,7 +826,12 @@ def _source_node( ), None, ) - return preserved or GraphNode( + if preserved is not None: + return preserved.model_copy( + update={"position": position}, + deep=True, + ) + return GraphNode( id=fallback_id, type="source.reference", label=source_name, @@ -1099,8 +1129,22 @@ def _column_name( return f"{prefix}{expression.name}" -def _position(index: int) -> GraphPosition: - return GraphPosition(x=80 + index * 220, y=180) +def _position(layer: int, *, y: float = LAYOUT_CENTER_Y) -> GraphPosition: + return GraphPosition( + x=LAYOUT_ORIGIN_X + layer * LAYOUT_LAYER_GAP, + y=y, + ) + + +def _branch_positions(count: int) -> list[GraphPosition]: + top = max( + 40, + LAYOUT_CENTER_Y - ((count - 1) * LAYOUT_BRANCH_GAP / 2), + ) + return [ + _position(0, y=top + index * LAYOUT_BRANCH_GAP) + for index in range(count) + ] def _combine_and(expressions: list[exp.Expression]) -> exp.Expression: diff --git a/tests/test_graph_and_sql.py b/tests/test_graph_and_sql.py index e2938d0..d82751b 100644 --- a/tests/test_graph_and_sql.py +++ b/tests/test_graph_and_sql.py @@ -13,8 +13,10 @@ from govoplan_dataflow.backend.schemas import ( GraphEdge, GraphNode, GraphPosition, + PipelineDraftRequest, PipelineGraph, ) +from govoplan_dataflow.backend.service import compile_sql_draft from govoplan_dataflow.backend.sql_compiler import SqlCompilationError, compile_sql, render_sql @@ -301,6 +303,14 @@ class DataflowGraphAndSqlTests(unittest.TestCase): ["source.inline", "source.inline", "combine.union", "output"], [node.type for node in roundtrip.nodes], ) + self.assertEqual( + [60, 60, 300, 540], + [node.position.x for node in roundtrip.nodes], + ) + self.assertEqual( + 120, + roundtrip.nodes[1].position.y - roundtrip.nodes[0].position.y, + ) self.assertEqual(result.rows, execute_preview(roundtrip, row_limit=100).rows) def test_legacy_composite_edge_ids_remain_readable(self) -> None: @@ -310,6 +320,110 @@ class DataflowGraphAndSqlTests(unittest.TestCase): self.assertEqual(edge_id, edge.id) + def test_sql_roundtrip_preserves_layout_for_compatible_topology(self) -> None: + compiled, sql_text, _ = compile_sql( + """ + SELECT * + FROM monthly_files + WHERE status = 'open' + ORDER BY amount DESC + """, + source_nodes=[inline_source()], + ) + id_map = { + node.id: f"custom-node-{index}" + for index, node in enumerate(compiled.nodes, start=1) + } + reference = compiled.model_copy( + update={ + "nodes": [ + node.model_copy( + update={ + "id": id_map[node.id], + "label": f"Custom {index}", + "position": GraphPosition( + x=111 + index * 137, + y=73 + index * 41, + ), + }, + deep=True, + ) + for index, node in enumerate(compiled.nodes, start=1) + ], + "edges": [ + edge.model_copy( + update={ + "id": f"custom-edge-{index}", + "source": id_map[edge.source], + "target": id_map[edge.target], + }, + deep=True, + ) + for index, edge in enumerate(compiled.edges, start=1) + ], + }, + deep=True, + ) + reference = reference.model_copy( + update={ + "nodes": list(reversed(reference.nodes)), + "edges": list(reversed(reference.edges)), + }, + deep=True, + ) + + response = compile_sql_draft( + PipelineDraftRequest( + graph=reference, + sql_text=sql_text.replace("'open'", "'closed'"), + ) + ) + + self.assertTrue(response.valid) + self.assertIsNotNone(response.graph) + assert response.graph is not None + self.assertEqual( + [ + (node.id, node.label, node.position) + for node in reference.nodes + ], + [ + (node.id, node.label, node.position) + for node in response.graph.nodes + ], + ) + self.assertEqual( + [edge.id for edge in reference.edges], + [edge.id for edge in response.graph.edges], + ) + filter_node = next( + node + for node in response.graph.nodes + if node.type == "filter" + ) + self.assertEqual("closed", filter_node.config["value"]) + + def test_changed_sql_topology_uses_compact_layer_layout(self) -> None: + reference, _, _ = compile_sql( + "SELECT * FROM monthly_files WHERE status = 'open'", + source_nodes=[inline_source()], + ) + + response = compile_sql_draft( + PipelineDraftRequest( + graph=reference, + sql_text="SELECT * FROM monthly_files", + ) + ) + + self.assertTrue(response.valid) + self.assertIsNotNone(response.graph) + assert response.graph is not None + self.assertEqual( + [60, 300], + [node.position.x for node in response.graph.nodes], + ) + def test_connector_source_uses_bounded_resolver_and_records_lineage(self) -> None: source = GraphNode( id="connector", diff --git a/webui/scripts/test-dataflow-page-structure.mjs b/webui/scripts/test-dataflow-page-structure.mjs index 1209071..7e80242 100644 --- a/webui/scripts/test-dataflow-page-structure.mjs +++ b/webui/scripts/test-dataflow-page-structure.mjs @@ -27,6 +27,14 @@ const checks = [ && !css.includes(".dataflow-alerts"), "central floating dismissible alerts" ], + [ + !css.includes(".dataflow-results-toolbar .segmented-control"), + "ungapped shared results segmented control" + ], + [ + page.includes("response.graph.nodes.some((node) => node.id === selectedNodeId)"), + "selected node preservation across SQL roundtrips" + ], [page.includes("useUnsavedDraftGuard"), "unsaved-change guard"], [canvas.includes("application/x-govoplan-dataflow-node"), "palette drop handling"], [canvas.includes("isValidConnection"), "edge validation"], diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index 0061810..1287fe7 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -397,7 +397,12 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings sqlText: response.sql_text ?? draft.sqlText, editorMode: switchToGraph ? "graph" : "sql" }); - setSelectedNodeId(response.graph.nodes[0]?.id ?? null); + setSelectedNodeId( + selectedNodeId + && response.graph.nodes.some((node) => node.id === selectedNodeId) + ? selectedNodeId + : response.graph.nodes[0]?.id ?? null + ); setSuccess("SQL compiled into the pipeline graph."); return true; } catch (compileError) { diff --git a/webui/src/styles/dataflow.css b/webui/src/styles/dataflow.css index ce9843b..64083ad 100644 --- a/webui/src/styles/dataflow.css +++ b/webui/src/styles/dataflow.css @@ -66,8 +66,7 @@ .dataflow-toolbar-actions, .dataflow-command-bar, -.dataflow-identity-fields, -.dataflow-results-toolbar .segmented-control { +.dataflow-identity-fields { display: flex; align-items: center; gap: 7px;