From 8724591bf936e19b1287ff4b3a8cf06b20902c01 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 31 Jul 2026 17:18:26 +0200 Subject: [PATCH] Complete SQL diagnostics and editor accessibility --- src/govoplan_dataflow/backend/schemas.py | 10 + src/govoplan_dataflow/backend/sql_compiler.py | 453 +++++++++++++++--- tests/test_graph_and_sql.py | 42 ++ webui/src/api/dataflow.ts | 8 + .../src/features/dataflow/DataflowCanvas.tsx | 27 ++ webui/src/features/dataflow/DataflowPage.tsx | 8 +- webui/src/styles/dataflow.css | 22 +- 7 files changed, 494 insertions(+), 76 deletions(-) diff --git a/src/govoplan_dataflow/backend/schemas.py b/src/govoplan_dataflow/backend/schemas.py index dccacd6..5d0895c 100644 --- a/src/govoplan_dataflow/backend/schemas.py +++ b/src/govoplan_dataflow/backend/schemas.py @@ -70,12 +70,22 @@ class PipelineGraph(BaseModel): edges: list[GraphEdge] = Field(default_factory=list, max_length=200) +class DiagnosticSourceLocation(BaseModel): + start_line: int = Field(ge=1) + start_column: int = Field(ge=1) + end_line: int = Field(ge=1) + end_column: int = Field(ge=1) + start_offset: int | None = Field(default=None, ge=0) + end_offset: int | None = Field(default=None, ge=0) + + class DataflowDiagnostic(BaseModel): severity: DiagnosticSeverity code: str message: str node_id: str | None = None field: str | None = None + source_location: DiagnosticSourceLocation | None = None class PipelineRevisionResponse(BaseModel): diff --git a/src/govoplan_dataflow/backend/sql_compiler.py b/src/govoplan_dataflow/backend/sql_compiler.py index 31a0fa0..a16d9b1 100644 --- a/src/govoplan_dataflow/backend/sql_compiler.py +++ b/src/govoplan_dataflow/backend/sql_compiler.py @@ -1,5 +1,6 @@ from __future__ import annotations +from contextvars import ContextVar from dataclasses import dataclass, field from typing import Any, Callable, Iterable @@ -12,6 +13,7 @@ from govoplan_dataflow.backend.expressions import parse_expression from govoplan_dataflow.backend.operator_registry import OPERATOR_REGISTRY from govoplan_dataflow.backend.schemas import ( DataflowDiagnostic, + DiagnosticSourceLocation, GraphEdge, GraphNode, GraphPosition, @@ -31,6 +33,18 @@ LAYOUT_LAYER_GAP = 240 LAYOUT_BRANCH_GAP = 120 +@dataclass(frozen=True) +class _SqlDiagnosticContext: + source_text: str + root_expression: exp.Expression | None = None + + +_SQL_DIAGNOSTIC_CONTEXT: ContextVar[_SqlDiagnosticContext | None] = ContextVar( + "dataflow_sql_diagnostic_context", + default=None, +) + + @dataclass class _SqlRenderState: where_conditions: list[exp.Expression] = field(default_factory=list) @@ -102,48 +116,60 @@ def compile_sql( *, source_nodes: Iterable[GraphNode] = (), ) -> tuple[PipelineGraph, str, list[DataflowDiagnostic]]: - query, union_expression = _parse_sql_query(sql_text) - from_clause, joins = _validated_source_clause( - query, - union_expression=union_expression, + source_text = sql_text + context_token = _SQL_DIAGNOSTIC_CONTEXT.set( + _SqlDiagnosticContext(source_text=source_text) ) - state = _initialize_compile_sources( - from_clause, - joins=joins, - union_expression=union_expression, - source_nodes=list(source_nodes), - ) - _append_filter_nodes(state, query) - projection = _projection_plan( - query, - qualifier_prefixes=state.qualifier_prefixes, - ) - _append_projection_node(state, query, projection) - _append_distinct_node(state, query, projection) - _append_sort_node(state, query) - _append_limit_node(state, query) - state.append_transform( - GraphNode( - id="output", - type="output", - label="Preview output", - position=state.next_transform_position(), - config={}, + try: + query, union_expression = _parse_sql_query(source_text) + _SQL_DIAGNOSTIC_CONTEXT.set( + _SqlDiagnosticContext( + source_text=source_text, + root_expression=union_expression or query, + ) ) - ) - graph = PipelineGraph(nodes=state.nodes, edges=state.edges) - diagnostics = validate_graph(graph) - if any(item.severity == "error" for item in diagnostics): - raise SqlCompilationError(diagnostics) - return graph, query.sql(dialect="duckdb", pretty=True), diagnostics + from_clause, joins = _validated_source_clause( + query, + union_expression=union_expression, + ) + state = _initialize_compile_sources( + from_clause, + joins=joins, + union_expression=union_expression, + source_nodes=list(source_nodes), + ) + _append_filter_nodes(state, query) + projection = _projection_plan( + query, + qualifier_prefixes=state.qualifier_prefixes, + ) + _append_projection_node(state, query, projection) + _append_distinct_node(state, query, projection) + _append_sort_node(state, query) + _append_limit_node(state, query) + state.append_transform( + GraphNode( + id="output", + type="output", + label="Preview output", + position=state.next_transform_position(), + config={}, + ) + ) + graph = PipelineGraph(nodes=state.nodes, edges=state.edges) + diagnostics = validate_graph(graph) + if any(item.severity == "error" for item in diagnostics): + raise SqlCompilationError(diagnostics) + return graph, query.sql(dialect="duckdb", pretty=True), diagnostics + finally: + _SQL_DIAGNOSTIC_CONTEXT.reset(context_token) def _parse_sql_query(sql_text: str) -> tuple[exp.Select, exp.Union | None]: - source_text = sql_text.strip() - if not source_text: + if not sql_text.strip(): raise SqlCompilationError([_sql_error("sql.empty", "Enter a SELECT query.")]) try: - statements = sqlglot.parse(source_text, read="duckdb") + statements = sqlglot.parse(sql_text, read="duckdb") except ParseError as exc: raise _parse_error(exc) from exc if len(statements) != 1 or not isinstance( @@ -155,6 +181,7 @@ def _parse_sql_query(sql_text: str) -> tuple[exp.Select, exp.Union | None]: _sql_error( "sql.select_only", "Dataflow SQL accepts exactly one SELECT or UNION BY NAME statement.", + expression=statements[0] if statements else None, ) ] ) @@ -172,6 +199,7 @@ def _parse_error(exc: ParseError) -> SqlCompilationError: _sql_error( "sql.parse", f"SQL could not be parsed{location}: {description}", + source_location=_parse_error_location(detail), ) ] ) @@ -325,12 +353,17 @@ def _initialize_tabular_sources( def _validate_logical_tables(tables: list[exp.Table]) -> None: - if any(table.catalog or table.db for table in tables): + qualified_table = next( + (table for table in tables if table.catalog or table.db), + None, + ) + if qualified_table is not None: raise SqlCompilationError( [ _sql_error( "sql.qualified_source", "Use logical source names without a catalog or schema.", + expression=qualified_table, ) ] ) @@ -477,6 +510,7 @@ def _projection_item( _sql_error( "sql.select_expression", "SELECT supports columns and COUNT/SUM/AVG/MIN/MAX only.", + expression=item, ) ] ) @@ -494,25 +528,55 @@ def _validate_projection_plan( ) -> None: if plan.saw_star and len(query.expressions) != 1: raise SqlCompilationError( - [_sql_error("sql.star_mix", "SELECT * cannot be mixed with other expressions in this dialect.")] + [ + _sql_error( + "sql.star_mix", + "SELECT * cannot be mixed with other expressions in this dialect.", + expression=query.expressions[0], + ) + ] ) if not (plan.saw_aggregate or plan.group_by): return if plan.saw_star: raise SqlCompilationError( - [_sql_error("sql.aggregate_star", "SELECT * cannot be grouped.")] + [ + _sql_error( + "sql.aggregate_star", + "SELECT * cannot be grouped.", + expression=query.expressions[0], + ) + ] ) if any(field["column"] not in plan.group_by for field in plan.fields): raise SqlCompilationError( - [_sql_error("sql.grouping", "Every non-aggregate SELECT column must appear in GROUP BY.")] + [ + _sql_error( + "sql.grouping", + "Every non-aggregate SELECT column must appear in GROUP BY.", + expression=query.args.get("group") or query, + ) + ] ) if any(field["alias"] != field["column"] for field in plan.fields): raise SqlCompilationError( - [_sql_error("sql.group_alias", "Aliases for GROUP BY columns are not supported yet.")] + [ + _sql_error( + "sql.group_alias", + "Aliases for GROUP BY columns are not supported yet.", + expression=query.args.get("group") or query, + ) + ] ) if not plan.aggregates: raise SqlCompilationError( - [_sql_error("sql.aggregate_required", "GROUP BY requires at least one aggregate in this dialect.")] + [ + _sql_error( + "sql.aggregate_required", + "GROUP BY requires at least one aggregate in this dialect.", + expression=query.args.get("group") or query, + ) + ] ) @@ -555,7 +619,13 @@ def _append_distinct_node( return if plan.saw_aggregate or plan.group_by: raise SqlCompilationError( - [_sql_error("sql.distinct_group", "DISTINCT cannot be combined with aggregation yet.")] + [ + _sql_error( + "sql.distinct_group", + "DISTINCT cannot be combined with aggregation yet.", + expression=query.args.get("distinct") or query, + ) + ] ) state.append_transform( GraphNode( @@ -597,7 +667,13 @@ def _order_field( ) -> dict[str, str]: if not isinstance(item, exp.Ordered): raise SqlCompilationError( - [_sql_error("sql.order", "Unsupported ORDER BY expression.")] + [ + _sql_error( + "sql.order", + "Unsupported ORDER BY expression.", + expression=item, + ) + ] ) return { "column": _column_name( @@ -628,17 +704,35 @@ def _append_limit_node(state: _SqlCompileState, query: exp.Select) -> None: def _limit_count(expression: exp.Expression | None) -> int: if not isinstance(expression, exp.Literal) or expression.is_string: raise SqlCompilationError( - [_sql_error("sql.limit", "LIMIT must be a positive integer.")] + [ + _sql_error( + "sql.limit", + "LIMIT must be a positive integer.", + expression=expression, + ) + ] ) try: count = int(expression.this) except (TypeError, ValueError) as exc: raise SqlCompilationError( - [_sql_error("sql.limit", "LIMIT must be a positive integer.")] + [ + _sql_error( + "sql.limit", + "LIMIT must be a positive integer.", + expression=expression, + ) + ] ) from exc if not 1 <= count <= 100_000: raise SqlCompilationError( - [_sql_error("sql.limit_range", "LIMIT must be between 1 and 100,000.")] + [ + _sql_error( + "sql.limit_range", + "LIMIT must be between 1 and 100,000.", + expression=expression, + ) + ] ) return count @@ -985,6 +1079,7 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: _sql_error( "sql.union_by_name", "Use UNION BY NAME so SQL matches Dataflow column-name append semantics.", + expression=expression, ) ] ) @@ -994,6 +1089,9 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: _sql_error( "sql.union_branch_order", "Apply ORDER BY or LIMIT after the complete UNION BY NAME.", + expression=expression.args.get("order") + or expression.args.get("limit") + or expression, ) ] ) @@ -1003,7 +1101,13 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: return if not isinstance(expression, exp.Select): raise SqlCompilationError( - [_sql_error("sql.union_branch", "UNION BY NAME branches must be SELECT statements.")] + [ + _sql_error( + "sql.union_branch", + "UNION BY NAME branches must be SELECT statements.", + expression=expression, + ) + ] ) branches.append(expression) @@ -1034,6 +1138,7 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: _sql_error( "sql.union_branch_shape", "UNION BY NAME inputs must be direct SELECT * source reads.", + expression=branch, ) ] ) @@ -1043,6 +1148,7 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: _sql_error( "sql.union_branch_projection", "UNION BY NAME inputs must select all source columns.", + expression=branch.expressions[0] if branch.expressions else branch, ) ] ) @@ -1050,11 +1156,23 @@ def _union_tables(union_expression: exp.Union) -> tuple[list[exp.Table], str]: 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.")] + [ + _sql_error( + "sql.union_source", + "Each UNION BY NAME input needs a logical source.", + expression=branch, + ) + ] ) if table.catalog or table.db: raise SqlCompilationError( - [_sql_error("sql.qualified_source", "Use logical source names without a catalog or schema.")] + [ + _sql_error( + "sql.qualified_source", + "Use logical source names without a catalog or schema.", + expression=table, + ) + ] ) tables.append(table) return tables, "distinct" if modes == {True} else "all" @@ -1074,15 +1192,35 @@ def _reject_unsupported_query_shape( "locks": "locking clauses", } for key, label in unsupported_args.items(): - if query.args.get(key): + unsupported = query.args.get(key) + if unsupported: raise SqlCompilationError( - [_sql_error("sql.unsupported_clause", f"{label} are not supported by the first Dataflow dialect.")] + [ + _sql_error( + "sql.unsupported_clause", + f"{label} are not supported by the first Dataflow dialect.", + expression=unsupported, + ) + ] ) - 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.")]) + unsupported_subquery = next( + ( + subquery + for subquery in query.find_all(exp.Subquery) + if subquery is not allowed_subquery + ), + None, + ) + if unsupported_subquery is not None: + raise SqlCompilationError( + [ + _sql_error( + "sql.subquery", + "Subqueries are not supported by the first Dataflow dialect.", + expression=unsupported_subquery, + ) + ] + ) def _source_node( @@ -1142,6 +1280,7 @@ def _join_config( "JOIN supports INNER, LEFT, RIGHT, FULL, SEMI, " "or ANTI joins only." ), + expression=join, ) ] ) @@ -1151,6 +1290,7 @@ def _join_config( _sql_error( "sql.join_type", "SEMI and ANTI joins cannot use a side qualifier.", + expression=join, ) ] ) @@ -1161,19 +1301,37 @@ def _join_config( ) if not join_type: raise SqlCompilationError( - [_sql_error("sql.join_type", "OUTER JOIN requires LEFT, RIGHT, or FULL.")] + [ + _sql_error( + "sql.join_type", + "OUTER JOIN requires LEFT, RIGHT, or FULL.", + expression=join, + ) + ] ) on_expression = join.args.get("on") if on_expression is None: raise SqlCompilationError( - [_sql_error("sql.join_condition", "JOIN requires an ON key comparison.")] + [ + _sql_error( + "sql.join_condition", + "JOIN requires an ON key comparison.", + expression=join, + ) + ] ) left_qualifiers = _table_qualifiers(left_table) right_qualifiers = _table_qualifiers(right_table) if left_qualifiers & right_qualifiers: raise SqlCompilationError( - [_sql_error("sql.join_alias", "Joined sources need distinct names or aliases.")] + [ + _sql_error( + "sql.join_alias", + "Joined sources need distinct names or aliases.", + expression=join, + ) + ] ) left_keys: list[str] = [] right_keys: list[str] = [] @@ -1184,7 +1342,13 @@ def _join_config( or not isinstance(condition.expression, exp.Column) ): raise SqlCompilationError( - [_sql_error("sql.join_condition", "JOIN ON accepts equality comparisons between source columns.")] + [ + _sql_error( + "sql.join_condition", + "JOIN ON accepts equality comparisons between source columns.", + expression=condition, + ) + ] ) first = condition.this second = condition.expression @@ -1202,6 +1366,7 @@ def _join_config( _sql_error( "sql.join_qualification", "Every JOIN key must qualify one left and one right source column.", + expression=condition, ) ] ) @@ -1236,7 +1401,15 @@ def _flatten_and(expression: exp.Expression) -> list[exp.Expression]: if isinstance(expression, exp.And): return [*_flatten_and(expression.this), *_flatten_and(expression.expression)] if isinstance(expression, exp.Or): - raise SqlCompilationError([_sql_error("sql.or", "OR conditions are not supported yet; use separate pipelines.")]) + raise SqlCompilationError( + [ + _sql_error( + "sql.or", + "OR conditions are not supported yet; use separate pipelines.", + expression=expression, + ) + ] + ) return [expression] @@ -1291,7 +1464,13 @@ def _condition_config( value = _literal_value(expression.expression) if not isinstance(value, str) or not (value.startswith("%") and value.endswith("%")): raise SqlCompilationError( - [_sql_error("sql.like", "LIKE is supported only as a contains pattern: LIKE '%value%'.")] + [ + _sql_error( + "sql.like", + "LIKE is supported only as a contains pattern: LIKE '%value%'.", + expression=expression, + ) + ] ) return { "column": _column_name( @@ -1303,7 +1482,13 @@ def _condition_config( "value": value[1:-1], } raise SqlCompilationError( - [_sql_error("sql.condition", "WHERE supports simple column comparisons joined with AND.")] + [ + _sql_error( + "sql.condition", + "WHERE supports simple column comparisons joined with AND.", + expression=expression, + ) + ] ) @@ -1785,7 +1970,15 @@ def _literal_value(expression: exp.Expression) -> Any: if isinstance(expression, exp.Boolean): return bool(expression.this) if not isinstance(expression, exp.Literal): - raise SqlCompilationError([_sql_error("sql.literal", "Comparisons require a literal value.")]) + raise SqlCompilationError( + [ + _sql_error( + "sql.literal", + "Comparisons require a literal value.", + expression=expression, + ) + ] + ) if expression.is_string: return str(expression.this) text = str(expression.this) @@ -1795,7 +1988,15 @@ def _literal_value(expression: exp.Expression) -> Any: try: return float(text) except ValueError as exc: - raise SqlCompilationError([_sql_error("sql.literal", f"Unsupported literal {text!r}.")]) from exc + raise SqlCompilationError( + [ + _sql_error( + "sql.literal", + f"Unsupported literal {text!r}.", + expression=expression, + ) + ] + ) from exc def _aggregate_config( @@ -1825,7 +2026,13 @@ def _aggregate_config( ) else: raise SqlCompilationError( - [_sql_error("sql.aggregate_argument", f"{function.upper()} requires a column or * argument.")] + [ + _sql_error( + "sql.aggregate_argument", + f"{function.upper()} requires a column or * argument.", + expression=argument, + ) + ] ) output_alias = alias or (f"{function}_all" if column == "*" else f"{function}_{column}") return {"function": function, "column": column, "alias": output_alias} @@ -1851,7 +2058,13 @@ def _column_name( ) -> str: if not isinstance(expression, exp.Column): raise SqlCompilationError( - [_sql_error("sql.column", f"{context} accepts column names only.")] + [ + _sql_error( + "sql.column", + f"{context} accepts column names only.", + expression=expression, + ) + ] ) if not expression.table and qualifier_prefixes is not None: raise SqlCompilationError( @@ -1859,6 +2072,7 @@ def _column_name( _sql_error( "sql.join_column_qualification", f"{context} columns must be source-qualified when a JOIN is present.", + expression=expression, ) ] ) @@ -1866,12 +2080,24 @@ def _column_name( return expression.name if qualifier_prefixes is None: raise SqlCompilationError( - [_sql_error("sql.column", f"{context} accepts unqualified column names only.")] + [ + _sql_error( + "sql.column", + f"{context} accepts unqualified column names only.", + expression=expression, + ) + ] ) prefix = qualifier_prefixes.get(expression.table.casefold()) if prefix is None: raise SqlCompilationError( - [_sql_error("sql.column_source", f"{context} references an unknown source qualifier.")] + [ + _sql_error( + "sql.column_source", + f"{context} references an unknown source qualifier.", + expression=expression, + ) + ] ) return f"{prefix}{expression.name}" @@ -1901,8 +2127,93 @@ def _combine_and(expressions: list[exp.Expression]) -> exp.Expression: return combined -def _sql_error(code: str, message: str) -> DataflowDiagnostic: - return DataflowDiagnostic(severity="error", code=code, message=message, field="sql_text") +def _sql_error( + code: str, + message: str, + *, + expression: exp.Expression | None = None, + source_location: DiagnosticSourceLocation | None = None, +) -> DataflowDiagnostic: + context = _SQL_DIAGNOSTIC_CONTEXT.get() + location_expression = expression or (context.root_expression if context else None) + return DataflowDiagnostic( + severity="error", + code=code, + message=message, + field="sql_text", + source_location=source_location or _expression_location(location_expression), + ) + + +def _expression_location( + expression: exp.Expression | None, +) -> DiagnosticSourceLocation | None: + context = _SQL_DIAGNOSTIC_CONTEXT.get() + if context is None or expression is None: + return None + offsets = [ + (int(node.meta["start"]), int(node.meta["end"])) + for node in expression.walk() + if node.meta.get("start") is not None and node.meta.get("end") is not None + ] + if not offsets: + return None + start_offset = min(item[0] for item in offsets) + end_offset = max(item[1] for item in offsets) + start_line, start_column = _line_and_column(context.source_text, start_offset) + end_line, end_column = _line_and_column(context.source_text, end_offset) + return DiagnosticSourceLocation( + start_line=start_line, + start_column=start_column, + end_line=end_line, + end_column=end_column, + start_offset=start_offset, + end_offset=end_offset, + ) + + +def _parse_error_location(detail: dict[str, Any]) -> DiagnosticSourceLocation | None: + context = _SQL_DIAGNOSTIC_CONTEXT.get() + line = detail.get("line") + column = detail.get("col") + if context is None or not isinstance(line, int) or not isinstance(column, int): + return None + highlight = str(detail.get("highlight") or "") + start_column = max(1, column - max(1, len(highlight)) + 1) + start_offset = _offset_for_line_and_column( + context.source_text, + line, + start_column, + ) + end_offset = min( + len(context.source_text) - 1, + start_offset + max(0, len(highlight) - 1), + ) + return DiagnosticSourceLocation( + start_line=line, + start_column=start_column, + end_line=line, + end_column=max(start_column, column), + start_offset=max(0, start_offset), + end_offset=max(0, end_offset), + ) + + +def _line_and_column(source_text: str, offset: int) -> tuple[int, int]: + bounded_offset = min(max(0, offset), max(0, len(source_text) - 1)) + line = source_text.count("\n", 0, bounded_offset) + 1 + last_newline = source_text.rfind("\n", 0, bounded_offset) + return line, bounded_offset - last_newline + + +def _offset_for_line_and_column(source_text: str, line: int, column: int) -> int: + lines = source_text.splitlines(keepends=True) + if not lines: + return 0 + bounded_line = min(max(1, line), len(lines)) + line_start = sum(len(item) for item in lines[: bounded_line - 1]) + line_content = lines[bounded_line - 1].rstrip("\r\n") + return line_start + min(max(0, column - 1), max(0, len(line_content) - 1)) def _node_sql_error(node_id: str, code: str, message: str) -> DataflowDiagnostic: diff --git a/tests/test_graph_and_sql.py b/tests/test_graph_and_sql.py index d82751b..311265c 100644 --- a/tests/test_graph_and_sql.py +++ b/tests/test_graph_and_sql.py @@ -138,6 +138,48 @@ class DataflowGraphAndSqlTests(unittest.TestCase): with self.subTest(sql=sql), self.assertRaises(SqlCompilationError): compile_sql(sql, source_nodes=[inline_source()]) + def test_sql_diagnostics_include_stable_source_locations(self) -> None: + cases = ( + ( + "SELECT *\nFROM monthly_files\nQUALIFY amount > 2", + "sql.unsupported_clause", + (3, 9, 3, 18, 36, 45), + ), + ( + "SELECT *\nFROM monthly_files\n" + "WHERE status = 'open' OR amount > 1", + "sql.or", + (3, 7, 3, 35, 34, 62), + ), + ( + "SELECT *\nFROM", + "sql.parse", + (2, 1, 2, 4, 9, 12), + ), + ) + + for sql, code, expected_location in cases: + with self.subTest(code=code), self.assertRaises(SqlCompilationError) as raised: + compile_sql(sql, source_nodes=[inline_source()]) + + diagnostic = raised.exception.diagnostics[0] + self.assertEqual(code, diagnostic.code) + self.assertEqual("sql_text", diagnostic.field) + self.assertIsNotNone(diagnostic.source_location) + location = diagnostic.source_location + assert location is not None + self.assertEqual( + expected_location, + ( + location.start_line, + location.start_column, + location.end_line, + location.end_column, + location.start_offset, + location.end_offset, + ), + ) + def test_compiles_renders_and_executes_two_source_join(self) -> None: sql = """ SELECT monthly_files.department, lookup.label diff --git a/webui/src/api/dataflow.ts b/webui/src/api/dataflow.ts index 2367b60..f63d25b 100644 --- a/webui/src/api/dataflow.ts +++ b/webui/src/api/dataflow.ts @@ -39,6 +39,14 @@ export type DataflowDiagnostic = { message: string; node_id?: string | null; field?: string | null; + source_location?: { + start_line: number; + start_column: number; + end_line: number; + end_column: number; + start_offset?: number | null; + end_offset?: number | null; + } | null; }; export type PipelineRevision = { diff --git a/webui/src/features/dataflow/DataflowCanvas.tsx b/webui/src/features/dataflow/DataflowCanvas.tsx index 158f776..a53162a 100644 --- a/webui/src/features/dataflow/DataflowCanvas.tsx +++ b/webui/src/features/dataflow/DataflowCanvas.tsx @@ -54,6 +54,7 @@ export default function DataflowCanvas({ onGraphChange, onSelectNode }: DataflowCanvasProps) { + const canvasRef = useRef(null); const [instance, setInstance] = useState | null>(null); const [proximityEdge, setProximityEdge] = useState(null); const [selectedEdgeId, setSelectedEdgeId] = useState(null); @@ -129,6 +130,24 @@ export default function DataflowCanvas({ }); }; + const restoreFocusAfterRemoval = (removedNodeIds: Set) => { + const focusedNodeId = document.activeElement + ?.closest(".react-flow__node") + ?.dataset.id; + if (!focusedNodeId || !removedNodeIds.has(focusedNodeId)) return; + const removedIndex = graph.nodes.findIndex((node) => node.id === focusedNodeId); + const remainingNodes = graph.nodes.filter((node) => !removedNodeIds.has(node.id)); + const nextNode = remainingNodes[Math.min(Math.max(0, removedIndex), remainingNodes.length - 1)]; + requestAnimationFrame(() => { + const nextElement = nextNode + ? canvasRef.current?.querySelector( + `.react-flow__node[data-id="${CSS.escape(nextNode.id)}"]` + ) + : null; + (nextElement ?? canvasRef.current)?.focus(); + }); + }; + const updateEdges = (nextEdges: Edge[]) => { onGraphChange({ ...graph, @@ -206,7 +225,10 @@ export default function DataflowCanvas({ return (
{ if (event.dataTransfer.types.includes("application/x-govoplan-dataflow-node")) { event.preventDefault(); @@ -224,6 +246,11 @@ export default function DataflowCanvas({ if (readOnly) return; const graphChanges = changes.filter((change) => change.type !== "dimensions"); if (graphChanges.length) { + restoreFocusAfterRemoval(new Set( + graphChanges + .filter((change) => change.type === "remove") + .map((change) => change.id) + )); updateNodes(applyNodeChanges(graphChanges, nodes)); } }} diff --git a/webui/src/features/dataflow/DataflowPage.tsx b/webui/src/features/dataflow/DataflowPage.tsx index b41d3be..1e6e8e7 100644 --- a/webui/src/features/dataflow/DataflowPage.tsx +++ b/webui/src/features/dataflow/DataflowPage.tsx @@ -2221,7 +2221,13 @@ function DiagnosticsPanel({ {item.code} {item.message} - {item.node_id ? {item.node_id} : null} + {item.node_id ? ( + {item.node_id} + ) : item.source_location ? ( + + {item.source_location.start_line}:{item.source_location.start_column} + + ) : null}
))} {nodeDiagnostics.map((item) => ( diff --git a/webui/src/styles/dataflow.css b/webui/src/styles/dataflow.css index 5fd8461..ebda6ed 100644 --- a/webui/src/styles/dataflow.css +++ b/webui/src/styles/dataflow.css @@ -359,6 +359,11 @@ grid-template-columns: minmax(0, 1fr) minmax(260px, 310px); } +.dataflow-canvas:focus-visible { + outline: 2px solid var(--accent); + outline-offset: -2px; +} + .dataflow-palette, .dataflow-inspector { min-width: 0; @@ -1120,7 +1125,7 @@ @media (max-width: 680px) { .dataflow-page { - height: calc(100vh - 94px); + height: calc(100dvh - 115px); } .dataflow-shell { @@ -1133,15 +1138,24 @@ border-bottom: var(--border-line); } + .dataflow-workspace-toolbar { + overflow-x: auto; + align-items: flex-start; + scrollbar-gutter: stable; + } + .dataflow-identity-fields, .dataflow-command-bar { - flex-wrap: wrap; + width: max-content; + min-width: max-content; + flex-wrap: nowrap; } .dataflow-name-input, .dataflow-description-input { - max-width: none; - min-width: 180px; + width: 220px; + min-width: 220px; + max-width: 220px; } .dataflow-editor {