From 74f1210a3295dc0c1b1608b4fded0550f30cdb91 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 17:54:28 +0200 Subject: [PATCH] Fix union SQL and selection previews --- README.md | 14 +- src/govoplan_dataflow/backend/schemas.py | 4 +- src/govoplan_dataflow/backend/sql_compiler.py | 419 ++++++++++++++---- tests/test_graph_and_sql.py | 21 + .../scripts/test-dataflow-page-structure.mjs | 21 + .../src/features/dataflow/DataflowCanvas.tsx | 5 +- webui/src/features/dataflow/DataflowPage.tsx | 123 +++-- .../features/dataflow/graphInteractions.ts | 4 + webui/src/styles/dataflow.css | 28 +- 9 files changed, 498 insertions(+), 141 deletions(-) diff --git a/README.md b/README.md index d8053ac..6e2be8c 100644 --- a/README.md +++ b/README.md @@ -45,18 +45,20 @@ fails visibly instead of silently changing a run. ## SQL And Preview Safety -The SQL workbench parses one `SELECT` statement into the canonical graph. The -dialect supports projection, aliases, filters, grouping, aggregate functions, -sorting, limits, `DISTINCT`, and one two-source equi-join. It rejects DDL, DML, -subqueries, arbitrary functions, file access, and unchecked pass-through -execution. +The SQL workbench parses one `SELECT` or column-aligned `UNION BY NAME` +statement into the canonical graph. The dialect supports projection, aliases, +filters, grouping, aggregate functions, sorting, limits, `DISTINCT`, append, +and one two-source equi-join. It rejects DDL, DML, arbitrary subqueries, +arbitrary functions, file access, and unchecked pass-through execution. Preview reads at most 250 rows per source and enforces time, intermediate-row, result-byte, graph-node, and response-row bounds. Saved previews record the pipeline revision, executor version, source fingerprints, node diagnostics, and output summary, but not source or result rows. A preview may return one explicitly requested intermediate node state under the same response bound; -that row payload remains ephemeral and is not added to run evidence. +that row payload remains ephemeral and is not added to run evidence. The WebUI +targets the selected graph node and supports remembered automatic or +manual-refresh preview modes. Saved revisions can also be started through the versioned `dataflow.runLifecycle` capability or the Run dialog. The first runner is diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index e29e86f..e72522c 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -45,7 +45,9 @@ class GraphNode(BaseModel): class GraphEdge(BaseModel): - id: str = Field(min_length=1, max_length=120, pattern=r"^[A-Za-z0-9_.:-]+$") + # Older WebUI releases composed edge IDs from both node IDs. Keep those + # definitions readable while new clients use short opaque edge IDs. + id: str = Field(min_length=1, max_length=255, pattern=r"^[A-Za-z0-9_.:-]+$") source: str = Field(min_length=1, max_length=100) target: str = Field(min_length=1, max_length=100) source_port: str = Field(default="output", min_length=1, max_length=80) diff --git a/src/govoplan_dataflow/backend/sql_compiler.py b/src/govoplan_dataflow/backend/sql_compiler.py index 6e069c5..259fc77 100644 --- a/src/govoplan_dataflow/backend/sql_compiler.py +++ b/src/govoplan_dataflow/backend/sql_compiler.py @@ -40,15 +40,30 @@ def compile_sql( raise SqlCompilationError( [_sql_error("sql.parse", f"SQL could not be parsed{location}: {detail.get('description') or exc}")] ) from exc - if len(statements) != 1 or not isinstance(statements[0], exp.Select): + if len(statements) != 1 or not isinstance(statements[0], (exp.Select, exp.Union)): raise SqlCompilationError( - [_sql_error("sql.select_only", "Dataflow SQL accepts exactly one SELECT statement.")] + [ + _sql_error( + "sql.select_only", + "Dataflow SQL accepts exactly one SELECT or UNION BY NAME statement.", + ) + ] ) - query = statements[0] - _reject_unsupported_query_shape(query) + query, union_expression = _select_query(statements[0]) from_clause = query.args.get("from_") - if from_clause is None or not isinstance(from_clause.this, exp.Table): + union_subquery = ( + from_clause.this + if from_clause is not None + and isinstance(from_clause.this, exp.Subquery) + and isinstance(from_clause.this.this, exp.Union) + else None + ) + _reject_unsupported_query_shape(query, allowed_subquery=union_subquery) + if from_clause is None or ( + union_expression is None + and not isinstance(from_clause.this, exp.Table) + ): raise SqlCompilationError( [_sql_error("sql.source_required", "SELECT requires a logical tabular source.")] ) @@ -57,81 +72,135 @@ def compile_sql( raise SqlCompilationError( [_sql_error("sql.join_count", "Dataflow SQL currently supports one two-source join.")] ) - left_table = from_clause.this - right_table = joins[0].this if joins else None - if right_table is not None and not isinstance(right_table, exp.Table): - raise SqlCompilationError( - [_sql_error("sql.join_source", "JOIN requires a logical tabular source.")] - ) - tables = [left_table, *([right_table] if isinstance(right_table, exp.Table) else [])] - for table in tables: - if table.catalog or table.db: - raise SqlCompilationError( - [_sql_error("sql.qualified_source", "Use logical source names without a catalog or schema.")] - ) - source_node_list = list(source_nodes) - left_source = _source_node( - 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), - ) - nodes = [left_source] + nodes: list[GraphNode] = [] edges: list[GraphEdge] = [] - previous_node_id = left_source.id qualifier_prefixes: dict[str, str] | None = None - if isinstance(right_table, exp.Table): - right_source = _source_node( - right_table, - source_node_list, - fallback_id="source-right", - position=GraphPosition(x=60, y=280), - ) - if right_source.id == left_source.id: + previous_node_id: str + edge_sequence = 0 + + def next_edge_id() -> str: + nonlocal edge_sequence + edge_sequence += 1 + return f"edge-{edge_sequence}" + + if union_expression is not None: + if joins: raise SqlCompilationError( - [_sql_error("sql.source_identity", "Joined sources must use different graph nodes.")] + [_sql_error("sql.union_join", "JOIN outside UNION BY NAME is not supported.")] ) - right_prefix = f"{right_table.alias_or_name}_" - join_config = _join_config(joins[0], left_table=left_table, right_table=right_table) - join_config["right_prefix"] = right_prefix - join_node = GraphNode( - id="join", - type="combine.join", - label=f"Join {left_table.name} and {right_table.name}", - position=GraphPosition(x=300, y=200), - config=join_config, + union_tables, union_mode = _union_tables(union_expression) + 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)), + ) + if any(existing.id == source.id for existing in nodes): + raise SqlCompilationError( + [ + _sql_error( + "sql.source_identity", + "UNION inputs must use different graph nodes.", + ) + ] + ) + nodes.append(source) + union_node = GraphNode( + id="union", + type="combine.union", + label="Append rows", + position=GraphPosition(x=300, y=80 + ((len(nodes) - 1) * 70)), + config={"mode": union_mode}, ) - nodes.extend((right_source, join_node)) + nodes.append(union_node) edges.extend( - ( - GraphEdge( - id=f"edge-{left_source.id}-{join_node.id}-left", - source=left_source.id, - target=join_node.id, - target_port="left", - ), - GraphEdge( - id=f"edge-{right_source.id}-{join_node.id}-right", - source=right_source.id, - target=join_node.id, - target_port="right", - ), + GraphEdge( + id=next_edge_id(), + source=source.id, + target=union_node.id, ) + for source in nodes + if source.type.startswith("source.") ) - previous_node_id = join_node.id - qualifier_prefixes = _join_qualifier_prefixes( + previous_node_id = union_node.id + else: + left_table = from_clause.this + if not isinstance(left_table, exp.Table): + raise SqlCompilationError( + [_sql_error("sql.source_required", "SELECT requires a logical tabular source.")] + ) + right_table = joins[0].this if joins else None + if right_table is not None and not isinstance(right_table, exp.Table): + raise SqlCompilationError( + [_sql_error("sql.join_source", "JOIN requires a logical tabular source.")] + ) + tables = [left_table, *([right_table] if isinstance(right_table, exp.Table) else [])] + for table in tables: + if table.catalog or table.db: + raise SqlCompilationError( + [_sql_error("sql.qualified_source", "Use logical source names without a catalog or schema.")] + ) + left_source = _source_node( left_table, - right_table, - right_prefix=right_prefix, + 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), ) + nodes.append(left_source) + previous_node_id = left_source.id + if isinstance(right_table, exp.Table): + right_source = _source_node( + right_table, + source_node_list, + fallback_id="source-right", + position=GraphPosition(x=60, y=280), + ) + if right_source.id == left_source.id: + raise SqlCompilationError( + [_sql_error("sql.source_identity", "Joined sources must use different graph nodes.")] + ) + right_prefix = f"{right_table.alias_or_name}_" + join_config = _join_config(joins[0], left_table=left_table, right_table=right_table) + join_config["right_prefix"] = right_prefix + join_node = GraphNode( + id="join", + type="combine.join", + label=f"Join {left_table.name} and {right_table.name}", + position=GraphPosition(x=300, y=200), + config=join_config, + ) + nodes.extend((right_source, join_node)) + edges.extend( + ( + GraphEdge( + id=next_edge_id(), + source=left_source.id, + target=join_node.id, + target_port="left", + ), + GraphEdge( + id=next_edge_id(), + source=right_source.id, + target=join_node.id, + target_port="right", + ), + ) + ) + previous_node_id = join_node.id + qualifier_prefixes = _join_qualifier_prefixes( + left_table, + right_table, + right_prefix=right_prefix, + ) def append_transform(node: GraphNode) -> None: nonlocal previous_node_id nodes.append(node) edges.append( GraphEdge( - id=f"edge-{previous_node_id}-{node.id}", + id=next_edge_id(), source=previous_node_id, target=node.id, ) @@ -322,17 +391,67 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: node_by_id = {node.id: node for node in graph.nodes} source_nodes = [node for node in graph.nodes if node.type.startswith("source.")] join_nodes = [node for node in graph.nodes if node.type == "combine.join"] + union_nodes = [node for node in graph.nodes if node.type == "combine.union"] if len(join_nodes) > 1: raise SqlCompilationError( [_sql_error("sql.join_count", "Only one two-source join can be rendered as SQL.")] ) + if len(union_nodes) > 1: + raise SqlCompilationError( + [_sql_error("sql.union_count", "Only one append node can be rendered as SQL.")] + ) + if join_nodes and union_nodes: + raise SqlCompilationError( + [_sql_error("sql.combine_count", "JOIN and UNION cannot yet be combined in Dataflow SQL.")] + ) - left_source: GraphNode + left_source: GraphNode | None = None right_source: GraphNode | None = None join_node = join_nodes[0] if join_nodes else None + union_node = union_nodes[0] if union_nodes else None + union_expression: exp.Expression | None = None + left_source_name: str | None = None right_prefix: str | None = None right_alias: str | None = None - if join_node is not None: + right_source_name: str | None = None + if union_node is not None: + inputs = graph_inputs_by_port(graph).get(union_node.id, {}) + union_sources = [ + node_by_id[source_id] + for source_id in inputs.get("input", []) + if source_id in node_by_id + ] + if ( + len(union_sources) != len(source_nodes) + or any(not node.type.startswith("source.") for node in union_sources) + or {node.id for node in union_sources} != {node.id for node in source_nodes} + ): + raise SqlCompilationError( + [ + _node_sql_error( + union_node.id, + "sql.union_shape", + "SQL rendering currently requires each append input to connect directly to a source.", + ) + ] + ) + source_names = [ + str(node.config.get("source_name", "")).strip() + for node in union_sources + ] + if any(not source_name for source_name in source_names): + raise SqlCompilationError( + [_sql_error("source.name_required", "Every source needs a logical SQL name.")] + ) + union_expression = exp.select("*").from_(exp.to_table(source_names[0])) + for source_name in source_names[1:]: + union_expression = exp.Union( + this=union_expression, + expression=exp.select("*").from_(exp.to_table(source_name)), + distinct=union_node.config.get("mode") == "distinct", + by_name=True, + ) + elif join_node is not None: inputs = graph_inputs_by_port(graph).get(join_node.id, {}) left_source = node_by_id[inputs["left"][0]] right_source = node_by_id[inputs["right"][0]] @@ -345,20 +464,25 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: [_sql_error("sql.source_count", "SQL rendering needs one source or one two-source join.")] ) - left_source_name = str(left_source.config.get("source_name", "")).strip() - right_source_name = ( - str(right_source.config.get("source_name", "")).strip() - if right_source is not None - else None - ) - if not left_source_name or (right_source is not None and not right_source_name): - raise SqlCompilationError( - [_sql_error("source.name_required", "Every source needs a logical SQL name.")] - ) - if right_alias and right_alias.casefold() == left_source_name.casefold(): - raise SqlCompilationError( - [_sql_error("sql.join_alias", "The right-column prefix conflicts with the left source name.")] + if union_node is None: + if left_source is None: + raise SqlCompilationError( + [_sql_error("sql.source_required", "SQL rendering needs a source.")] + ) + left_source_name = str(left_source.config.get("source_name", "")).strip() + right_source_name = ( + str(right_source.config.get("source_name", "")).strip() + if right_source is not None + else None ) + if not left_source_name or (right_source is not None and not right_source_name): + raise SqlCompilationError( + [_sql_error("source.name_required", "Every source needs a logical SQL name.")] + ) + if right_alias and right_alias.casefold() == left_source_name.casefold(): + raise SqlCompilationError( + [_sql_error("sql.join_alias", "The right-column prefix conflicts with the left source name.")] + ) def column_expression(name: str) -> exp.Column: if right_prefix and right_alias and name.startswith(right_prefix): @@ -368,7 +492,7 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: [_sql_error("sql.column", "A right-side column name is missing.")] ) return exp.column(right_name, table=right_alias) - if right_source is not None: + if right_source is not None and left_source_name: return exp.column(name, table=left_source_name) return exp.column(name) @@ -382,7 +506,7 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: for node_id in ordered: node = node_by_id[node_id] - if node.type.startswith("source.") or node.type == "combine.join": + if node.type.startswith("source.") or node.type in {"combine.join", "combine.union"}: continue if node.type == "filter": if selected: @@ -482,7 +606,16 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: [_node_sql_error(node.id, "sql.node_not_representable", f"{node.type!r} cannot be rendered as SQL.")] ) - query = exp.select(*select_expressions).from_(exp.to_table(left_source_name)) + if union_expression is not None: + query = exp.select(*select_expressions).from_( + union_expression.subquery("_unioned") + ) + elif left_source_name: + query = exp.select(*select_expressions).from_(exp.to_table(left_source_name)) + else: + raise SqlCompilationError( + [_sql_error("sql.source_required", "SQL rendering needs a source.")] + ) if join_node is not None and right_source_name and right_alias: join_conditions = [ exp.EQ( @@ -513,7 +646,124 @@ def render_sql(graph: PipelineGraph) -> tuple[str, list[DataflowDiagnostic]]: return query.sql(dialect="duckdb", pretty=True), diagnostics -def _reject_unsupported_query_shape(query: exp.Select) -> None: +def _select_query( + statement: exp.Select | exp.Union, +) -> tuple[exp.Select, exp.Union | None]: + if isinstance(statement, exp.Select): + from_clause = statement.args.get("from_") + if ( + from_clause is not None + and isinstance(from_clause.this, exp.Subquery) + and isinstance(from_clause.this.this, exp.Union) + ): + return statement, from_clause.this.this + return statement, None + + union_expression = statement.copy() + order = union_expression.args.get("order") + limit = union_expression.args.get("limit") + union_expression.set("order", None) + union_expression.set("limit", None) + query = exp.select("*").from_(union_expression.subquery("_unioned")) + if order is not None: + query.set("order", order.copy()) + if limit is not None: + query.set("limit", limit.copy()) + return query, union_expression + + +def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: + branches: list[exp.Select] = [] + modes: set[bool] = set() + + def collect(expression: exp.Expression) -> None: + if isinstance(expression, exp.Union): + if expression.args.get("by_name") is not True: + raise SqlCompilationError( + [ + _sql_error( + "sql.union_by_name", + "Use UNION BY NAME so SQL matches Dataflow column-name append semantics.", + ) + ] + ) + if expression.args.get("order") or expression.args.get("limit"): + raise SqlCompilationError( + [ + _sql_error( + "sql.union_branch_order", + "Apply ORDER BY or LIMIT after the complete UNION BY NAME.", + ) + ] + ) + modes.add(bool(expression.args.get("distinct"))) + collect(expression.this) + collect(expression.expression) + return + if not isinstance(expression, exp.Select): + raise SqlCompilationError( + [_sql_error("sql.union_branch", "UNION BY NAME branches must be SELECT statements.")] + ) + branches.append(expression) + + collect(union_expression) + if len(modes) != 1: + raise SqlCompilationError( + [ + _sql_error( + "sql.union_mode", + "A Dataflow append node cannot mix UNION and UNION ALL.", + ) + ] + ) + + tables: list[exp.Table] = [] + for branch in branches: + _reject_unsupported_query_shape(branch) + if ( + branch.args.get("joins") + or branch.args.get("where") + or branch.args.get("group") + or branch.args.get("order") + or branch.args.get("limit") + or branch.args.get("distinct") + ): + raise SqlCompilationError( + [ + _sql_error( + "sql.union_branch_shape", + "UNION BY NAME inputs must be direct SELECT * source reads.", + ) + ] + ) + if len(branch.expressions) != 1 or not isinstance(branch.expressions[0], exp.Star): + raise SqlCompilationError( + [ + _sql_error( + "sql.union_branch_projection", + "UNION BY NAME inputs must select all source columns.", + ) + ] + ) + from_clause = branch.args.get("from_") + table = from_clause.this if from_clause is not None else None + if not isinstance(table, exp.Table): + raise SqlCompilationError( + [_sql_error("sql.union_source", "Each UNION BY NAME input needs a logical source.")] + ) + if table.catalog or table.db: + raise SqlCompilationError( + [_sql_error("sql.qualified_source", "Use logical source names without a catalog or schema.")] + ) + tables.append(table) + return tables, "distinct" if modes == {True} else "all" + + +def _reject_unsupported_query_shape( + query: exp.Select, + *, + allowed_subquery: exp.Subquery | None = None, +) -> None: unsupported_args = { "with_": "WITH queries", "having": "HAVING", @@ -527,7 +777,10 @@ def _reject_unsupported_query_shape(query: exp.Select) -> None: raise SqlCompilationError( [_sql_error("sql.unsupported_clause", f"{label} are not supported by the first Dataflow dialect.")] ) - if any(True for _ in query.find_all(exp.Subquery)): + if any( + subquery is not allowed_subquery + for subquery in query.find_all(exp.Subquery) + ): raise SqlCompilationError([_sql_error("sql.subquery", "Subqueries are not supported by the first Dataflow dialect.")]) diff --git a/tests/test_graph_and_sql.py b/tests/test_graph_and_sql.py index 3e6cfa2..e2938d0 100644 --- a/tests/test_graph_and_sql.py +++ b/tests/test_graph_and_sql.py @@ -288,6 +288,27 @@ class DataflowGraphAndSqlTests(unittest.TestCase): result = execute_preview(graph, row_limit=100) self.assertEqual([{"id": 1}, {"id": 2}, {"id": 3}], result.rows) + rendered, diagnostics = render_sql(graph) + self.assertEqual([], diagnostics) + self.assertIn("UNION BY NAME", rendered) + roundtrip, normalized, diagnostics = compile_sql( + rendered, + source_nodes=[first, second], + ) + self.assertEqual(rendered, normalized) + self.assertEqual([], diagnostics) + self.assertEqual( + ["source.inline", "source.inline", "combine.union", "output"], + [node.type for node in roundtrip.nodes], + ) + self.assertEqual(result.rows, execute_preview(roundtrip, row_limit=100).rows) + + def test_legacy_composite_edge_ids_remain_readable(self) -> None: + edge_id = f"edge-{'a' * 100}-{'b' * 100}" + + edge = GraphEdge(id=edge_id, source="source", target="output") + + self.assertEqual(edge_id, edge.id) def test_connector_source_uses_bounded_resolver_and_records_lineage(self) -> None: source = GraphNode( diff --git a/webui/scripts/test-dataflow-page-structure.mjs b/webui/scripts/test-dataflow-page-structure.mjs index 2becaa8..1209071 100644 --- a/webui/scripts/test-dataflow-page-structure.mjs +++ b/webui/scripts/test-dataflow-page-structure.mjs @@ -11,6 +11,22 @@ const checks = [ [page.includes("Apply SQL"), "SQL workbench"], [page.includes("([]); const [nodeDiagnostics, setNodeDiagnostics] = useState([]); const [preview, setPreview] = useState(null); + const [autoPreview, setAutoPreview] = useState(readAutoPreviewPreference); const [resultOpen, setResultOpen] = useState(false); const [resultTab, setResultTab] = useState("preview"); const [deleteOpen, setDeleteOpen] = useState(false); @@ -120,6 +123,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings const [sources, setSources] = useState([]); const [sourceCatalogueAvailable, setSourceCatalogueAvailable] = useState(false); const [sourceCatalogueWritable, setSourceCatalogueWritable] = useState(false); + const previewRequestId = useRef(0); const dirty = Boolean(draft) && draftFingerprint(draft) !== draftFingerprint(savedDraft); const canWrite = hasScope(auth, "dataflow:pipeline:write") || hasScope(auth, "dataflow:pipeline:admin"); @@ -345,10 +349,12 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings }; const updateGraph = (graph: PipelineDraft["graph"]) => { + previewRequestId.current += 1; updateDraft({ graph }); setDiagnostics([]); setNodeDiagnostics([]); setPreview(null); + setWorking(false); }; const validate = async () => { @@ -426,13 +432,18 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings await applySql(true); }; - const runPreview = async (previewNodeId?: string) => { + const runPreview = useCallback(async ( + previewNodeId?: string, + quiet = false + ) => { if (!draft || !canPreview) return; const targetNodeId = previewNodeId ?? draft.graph.nodes.find((node) => node.type === "output")?.id; + const requestId = previewRequestId.current + 1; + previewRequestId.current = requestId; setWorking(true); setError(""); - setSuccess(""); + if (!quiet) setSuccess(""); try { const response = await previewDataflowPipeline(settings, !dirty && draft.id && draft.currentRevision ? { @@ -448,6 +459,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings preview_node_id: targetNodeId, row_limit: 100 }); + if (requestId !== previewRequestId.current) return; setPreview(response); setDiagnostics(response.diagnostics); setNodeDiagnostics(response.node_diagnostics); @@ -457,7 +469,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings ? "preview" : "diagnostics" ); - if (response.status === "succeeded" && response.node_preview) { + if (!quiet && response.status === "succeeded" && response.node_preview) { const node = draft.graph.nodes.find( (candidate) => candidate.id === response.node_preview?.node_id ); @@ -467,10 +479,38 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings ); } } catch (previewError) { + if (requestId !== previewRequestId.current) return; setError(apiErrorMessage(previewError)); } finally { - setWorking(false); + if (requestId === previewRequestId.current) setWorking(false); } + }, [canPreview, dirty, draft, settings]); + + useEffect(() => { + if ( + !autoPreview + || !selectedNodeId + || !canPreview + || draft?.editorMode !== "graph" + ) { + return; + } + const timeout = window.setTimeout(() => { + void runPreview(selectedNodeId, true); + }, 350); + return () => window.clearTimeout(timeout); + }, [autoPreview, canPreview, draft?.editorMode, runPreview, selectedNodeId]); + + const selectNode = useCallback((nodeId: string | null) => { + setSelectedNodeId(nodeId); + if (nodeId && nodeId !== preview?.node_preview?.node_id) { + setPreview(null); + } + }, [preview?.node_preview?.node_id]); + + const changeAutoPreview = (enabled: boolean) => { + setAutoPreview(enabled); + writeAutoPreviewPreference(enabled); }; const removePipeline = async () => { @@ -614,14 +654,23 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings + - {(error || success) ? ( -
- {error ? {error} : null} - {success ? {success} : null} -
+ {error ? ( + {error} + ) : null} + {success ? ( + {success} ) : null}
{draft.editorMode === "graph" ? ( @@ -742,7 +791,7 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings selectedNodeId={selectedNodeId} readOnly={!canEdit} onGraphChange={updateGraph} - onSelectNode={setSelectedNodeId} + onSelectNode={selectNode} /> ) : ( @@ -793,10 +842,9 @@ export default function DataflowPage({ settings, auth }: { settings: ApiSettings onTabChange={setResultTab} preview={preview} nodes={draft.graph.nodes} - working={working} diagnostics={diagnostics} nodeDiagnostics={nodeDiagnostics} - onPreviewNodeChange={(nodeId) => void runPreview(nodeId)} + selectedNodeId={selectedNodeId} onClose={() => setResultOpen(false)} /> ) : null} @@ -1872,24 +1920,23 @@ function ResultPanel({ onTabChange, preview, nodes, - working, diagnostics, nodeDiagnostics, - onPreviewNodeChange, + selectedNodeId, onClose }: { tab: ResultTab; onTabChange: (tab: ResultTab) => void; preview: PipelinePreview | null; nodes: PipelineGraphNode[]; - working: boolean; diagnostics: DataflowDiagnostic[]; nodeDiagnostics: NodePreviewDiagnostic[]; - onPreviewNodeChange: (nodeId: string) => void; + selectedNodeId: string | null; onClose: () => void; }) { const outputNodeId = nodes.find((node) => node.type === "output")?.id ?? ""; - const previewNodeId = preview?.node_preview?.node_id ?? outputNodeId; + const previewNodeId = preview?.node_preview?.node_id ?? selectedNodeId ?? outputNodeId; + const previewNode = nodes.find((node) => node.id === previewNodeId); const previewRows = preview?.node_preview?.total_rows ?? preview?.total_rows; return (
@@ -1904,19 +1951,11 @@ function ResultPanel({ value={tab} onChange={onTabChange} /> - {tab === "preview" && nodes.length ? ( - + {tab === "preview" && previewNode ? ( + + {previewNode.label} + {previewNode.type === "output" ? " (final output)" : ""} + ) : null}
@@ -2016,6 +2055,26 @@ function uniqueNodeExists(draft: PipelineDraft, type: string): boolean { return false; } +function readAutoPreviewPreference(): boolean { + if (typeof window === "undefined") return true; + try { + return window.localStorage.getItem(AUTO_PREVIEW_STORAGE_KEY) !== "manual"; + } catch { + return true; + } +} + +function writeAutoPreviewPreference(enabled: boolean): void { + try { + window.localStorage.setItem( + AUTO_PREVIEW_STORAGE_KEY, + enabled ? "auto" : "manual" + ); + } catch { + // The preference is non-essential when browser storage is unavailable. + } +} + function startPaletteDrag(event: DragEvent, type: string) { event.dataTransfer.setData("application/x-govoplan-dataflow-node", type); event.dataTransfer.effectAllowed = "copy"; diff --git a/webui/src/features/dataflow/graphInteractions.ts b/webui/src/features/dataflow/graphInteractions.ts index 5b91261..86e4c10 100644 --- a/webui/src/features/dataflow/graphInteractions.ts +++ b/webui/src/features/dataflow/graphInteractions.ts @@ -9,6 +9,10 @@ import type { DataflowFlowNode } from "./DataflowNode"; const PROXIMITY_GAP = 72; const PROXIMITY_EDGE_ID = "dataflow-proximity-preview"; +export function newGraphEdgeId(): string { + return `edge-${crypto.randomUUID()}`; +} + export function graphEdgesFromFlow(edges: Edge[]): PipelineGraph["edges"] { return edges .filter((edge) => edge.id !== PROXIMITY_EDGE_ID) diff --git a/webui/src/styles/dataflow.css b/webui/src/styles/dataflow.css index f09baca..ce9843b 100644 --- a/webui/src/styles/dataflow.css +++ b/webui/src/styles/dataflow.css @@ -343,20 +343,9 @@ gap: 5px; } -.dataflow-alerts { - position: absolute; - z-index: 12; - top: 66px; - right: 12px; - display: grid; - width: min(500px, calc(100% - 24px)); - gap: 6px; - pointer-events: none; -} - -.dataflow-alerts .alert { - pointer-events: auto; - box-shadow: var(--shadow-popover); +.dataflow-command-bar .toggle-switch-row { + min-height: 34px; + white-space: nowrap; } .dataflow-editor { @@ -923,9 +912,14 @@ flex-wrap: wrap; } -.dataflow-results-view-controls select { - width: clamp(150px, 22vw, 280px); - min-height: 30px; +.dataflow-preview-stage { + max-width: min(36vw, 320px); + overflow: hidden; + color: var(--text-strong); + font-size: 12px; + font-weight: 650; + text-overflow: ellipsis; + white-space: nowrap; } .dataflow-preview-table-wrap,