diff --git a/README.md b/README.md index 370c510..95e72f1 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,9 @@ nodes by purpose: | Group | Nodes | | --- | --- | | Load | Inline data, datasource | -| Combine | Append rows, join tables | +| Combine | Append rows, inner/outer/semi/anti joins | | Filter | Column filter, typed-expression filter, remove duplicates | -| Transform | Select, derive, typed expression, conversion, replacement, aggregate, sort, limit, reusable subflow | +| Transform | Select, derive, ordered calculations, typed expression, conversion, replacement, aggregate, partitioned rank, sort, limit, reusable subflow | | Quality | Quality rules, keyed reconciliation | | Output | Preview output | @@ -56,8 +56,9 @@ and one two-source equi-join. It rejects DDL, DML, arbitrary subqueries, arbitrary functions, file access, and unchecked pass-through execution. The typed-expression library is a separate allowlisted AST evaluator shared by expression filters and calculated columns. It supports literals, columns, -arithmetic, comparisons, boolean logic, `CASE`, safe casts, and a bounded -string/numeric function catalogue without Python evaluation or effectful SQL. +arithmetic, comparisons, boolean logic, `CASE`, safe casts, predicates, +date arithmetic/parts/formatting, and a bounded string/numeric function +catalogue without Python evaluation or effectful SQL. Node definitions, validators, schema propagators, preview executors, and SQL renderers are registered independently in the operator registry. Adding a node @@ -131,9 +132,11 @@ source inside the snapshot, parameter substitution is data-only, and nesting is bounded. This keeps completed run definitions reproducible even when the source template changes later. -The executable fixtures in `fixtures/golden` cover the monthly structured-file -reconciliation story and the sanctions-screening story with reviewable sample -inputs and exact expected outputs. +The executable fixtures in `fixtures/golden` cover monthly structured-file +reconciliation, sanctions screening, a HEICO-style current-status export, and +the set-based core of a RELE-style booking workflow with reviewable synthetic +inputs and exact expected outputs. The detailed source-flow assessment is in +[`docs/HEICO_RELE_ASSESSMENT.md`](docs/HEICO_RELE_ASSESSMENT.md). ## Development diff --git a/docs/HEICO_RELE_ASSESSMENT.md b/docs/HEICO_RELE_ASSESSMENT.md new file mode 100644 index 0000000..7d6d66c --- /dev/null +++ b/docs/HEICO_RELE_ASSESSMENT.md @@ -0,0 +1,111 @@ +# HEICO And RELE Dataflow Assessment + +This assessment uses the production-shaped SQL files supplied outside the +repositories: + +- `heico.sql`: Oracle SQL*Plus query, 285 lines +- `rele_sql`: PostgreSQL PL/pgSQL program, 4,223 lines + +The requested `rele.sql` filename does not exist; `rele_sql` is the matching +input. No production rows or source code are copied into this repository. The +executable contracts under `fixtures/golden` use synthetic data. + +## HEICO + +HEICO is predominantly a declarative relational flow and is a good fit for +Dataflow. Its relevant operations are: + +| Source operation | GovOPlaN representation | State | +| --- | --- | --- | +| CTE pipeline | Connected graph stages | Modelable; automatic CTE import is missing | +| Inner and left joins | Join node | Available | +| `NOT EXISTS` exclusion | Anti join | Available | +| Latest status per partition | Partitioned row number/rank | Available | +| Filters, grouping, aggregates | Filter and aggregate nodes | Available | +| `CASE` and calculated export fields | Ordered calculation block | Available | +| String/date normalization | Safe expression functions | Available for the used core functions | +| Correlated scalar lookup | Pre-aggregate plus join | Modelable, but not imported automatically | +| Oracle package function | Governed source projection/pushdown | Provider-specific work remains | +| CSV formatting/publication | Reporting or Files output target | Cross-module work remains | + +`fixtures/golden/heico-student-status` proves the critical shape: select the +latest status per student, remove excluded records with an anti join, calculate +conditional/export columns, and project the final result. + +The complete HEICO query can therefore be rebuilt as a governed graph today. +What is not yet available is one-click import of its Oracle dialect, automatic +decomposition of correlated lookups, and an Oracle provider contract for its +site-specific function. + +## RELE + +RELE is not one dataflow. It combines relational transformations with +procedural orchestration and effects: + +- temporary raw and staging tables plus indexes +- repeated ordered `UPDATE ... FROM` enrichment +- conditional branches and loops +- validation/error records +- inserts, updates, and deletes +- reconciliation and aggregate controls +- fixed-width and tabular output generation + +The correct GovOPlaN decomposition is: + +| Concern | Owning module | Current fit | +| --- | --- | --- | +| Immutable raw input and staging states | Datasources | Strong foundation | +| Set-based normalization, joins, calculations, aggregation | Dataflow | Strong after this slice | +| Ordered enrichments | Dataflow stages or reusable subflows | Modelable, manual decomposition | +| Validation and reconciliation | Dataflow quality/reconciliation | Available for single-output checks | +| User verification, correction, rerun, resumability | Workflow | Foundation exists; flow-specific handoffs remain | +| Database writes and cleanup | Governed Datasource output/effect capability | Missing explicit effect contract | +| Warning/error side streams | Multi-output Dataflow nodes | Missing | +| Fixed-width, CSV, and spreadsheet products | Reporting/Templates/Files | Missing integrated output profile | + +`fixtures/golden/rele-booking-transform` proves a representative set-based +slice: ordered calculated columns can normalize types, derive period and +conditional classifications, calculate signed amounts, validate required +fields, and aggregate the result. + +RELE should not be ported as a single opaque stored procedure. Its duplicated +booking/error paths should become reusable subflows, while Workflow owns the +run/verify/correct lifecycle. Database effects must remain separately +authorized and idempotent so a retry cannot repeat an uncertain write. + +## Added Building Blocks + +This assessment added: + +- ordered multi-column calculations, where later expressions can use columns + calculated earlier in the same block +- semi and anti joins +- partitioned `row_number`, `rank`, and `dense_rank` +- safe `LIKE`, `ILIKE`, `IN`, `BETWEEN`, regular-expression matching and + replacement +- padding, splitting, translation, concatenation, extrema, and `NULLIF` +- date parts, truncation, bounded interval arithmetic/difference, and temporal + formatting +- schema propagation, deterministic reference execution, typed IR inventory, + SQL rendering where representable, and editor controls for each new node + +The evaluator remains allowlisted. It does not execute arbitrary SQL, Python, +stored procedures, filesystem access, or network calls. + +## Remaining Work + +The next useful slices, in order, are: + +1. Import and decompose multi-CTE SQL, including window expressions and + correlated aggregate lookups. +2. Add reusable lookup/as-of join semantics for temporal master data. +3. Add multi-output nodes for valid, warning, and rejected rows. +4. Define explicit idempotent write/effect nodes against Datasource + capabilities, guarded by Policy and production approvals. +5. Add parameterized fixed-width/delimited output profiles in Reporting and + Templates, publishing through Files. +6. Add provider-side pushdown contracts for source-specific functions without + exposing unrestricted SQL execution. + +These gaps remain tracked by +[`govoplan-dataflow#17`](https://git.add-ideas.de/GovOPlaN/govoplan-dataflow/issues/17). diff --git a/fixtures/golden/README.md b/fixtures/golden/README.md index c66769e..79d9d7a 100644 --- a/fixtures/golden/README.md +++ b/fixtures/golden/README.md @@ -9,3 +9,8 @@ Each directory is an executable product contract: Source nodes use a fixture filename in `config.fixture`; the golden-flow test injects those rows before validation and bounded execution. + +Current contracts cover monthly reconciliation, sanctions screening, a +HEICO-style current-status export, and the set-based normalization core of a +RELE-style booking workflow. Fixtures are synthetic and contain no production +records. diff --git a/fixtures/golden/heico-student-status/expected-output.json b/fixtures/golden/heico-student-status/expected-output.json new file mode 100644 index 0000000..ba7a45e --- /dev/null +++ b/fixtures/golden/heico-student-status/expected-output.json @@ -0,0 +1,16 @@ +[ + { + "student_id": "S-100", + "full_name": "Ada Lovelace", + "birth_date": "10.12.2000", + "program": "CS", + "status_group": "current" + }, + { + "student_id": "S-300", + "full_name": "Katherine Johnson", + "birth_date": "26.08.2001", + "program": "MATH", + "status_group": "current" + } +] diff --git a/fixtures/golden/heico-student-status/graph.json b/fixtures/golden/heico-student-status/graph.json new file mode 100644 index 0000000..04097e8 --- /dev/null +++ b/fixtures/golden/heico-student-status/graph.json @@ -0,0 +1,130 @@ +{ + "schema_version": 1, + "nodes": [ + { + "id": "statuses", + "type": "source.inline", + "label": "Student statuses", + "position": {"x": 40, "y": 80}, + "config": { + "source_name": "student_statuses", + "fixture": "student-statuses.json", + "rows": [] + } + }, + { + "id": "exclusions", + "type": "source.inline", + "label": "Export exclusions", + "position": {"x": 520, "y": 300}, + "config": { + "source_name": "export_exclusions", + "fixture": "export-exclusions.json", + "rows": [] + } + }, + { + "id": "latest-rank", + "type": "window.rank", + "label": "Latest status per student", + "position": {"x": 280, "y": 80}, + "config": { + "method": "row_number", + "target_column": "status_rank", + "partition_by": ["student_id"], + "order_by": [ + {"column": "status_date", "direction": "desc"} + ] + } + }, + { + "id": "latest-only", + "type": "filter", + "label": "Keep latest status", + "position": {"x": 520, "y": 80}, + "config": { + "column": "status_rank", + "operator": "eq", + "value": 1 + } + }, + { + "id": "eligible-only", + "type": "combine.join", + "label": "Exclude ineligible students", + "position": {"x": 760, "y": 160}, + "config": { + "join_type": "anti", + "left_keys": ["student_id"], + "right_keys": ["student_id"], + "right_prefix": "excluded_" + } + }, + { + "id": "export-calculations", + "type": "calculate", + "label": "Calculate export fields", + "position": {"x": 1000, "y": 160}, + "config": { + "calculations": [ + { + "target_column": "full_name", + "expression": "trim(first_name) || ' ' || trim(last_name)", + "result_type": "string" + }, + { + "target_column": "birth_date_display", + "expression": "to_char(birth_date, 'DD.MM.YYYY')", + "result_type": "string" + }, + { + "target_column": "status_group", + "expression": "case when status in ('active', 'enrolled') then 'current' else 'inactive' end", + "result_type": "string" + } + ] + } + }, + { + "id": "export-fields", + "type": "select", + "label": "HEICO export", + "position": {"x": 1240, "y": 160}, + "config": { + "fields": [ + {"column": "student_id", "alias": "student_id"}, + {"column": "full_name", "alias": "full_name"}, + {"column": "birth_date_display", "alias": "birth_date"}, + {"column": "program", "alias": "program"}, + {"column": "status_group", "alias": "status_group"} + ] + } + }, + { + "id": "output", + "type": "output", + "label": "Export rows", + "position": {"x": 1480, "y": 160}, + "config": {} + } + ], + "edges": [ + {"id": "e1", "source": "statuses", "target": "latest-rank"}, + {"id": "e2", "source": "latest-rank", "target": "latest-only"}, + { + "id": "e3", + "source": "latest-only", + "target": "eligible-only", + "target_port": "left" + }, + { + "id": "e4", + "source": "exclusions", + "target": "eligible-only", + "target_port": "right" + }, + {"id": "e5", "source": "eligible-only", "target": "export-calculations"}, + {"id": "e6", "source": "export-calculations", "target": "export-fields"}, + {"id": "e7", "source": "export-fields", "target": "output"} + ] +} diff --git a/fixtures/golden/heico-student-status/inputs/export-exclusions.json b/fixtures/golden/heico-student-status/inputs/export-exclusions.json new file mode 100644 index 0000000..cff4c14 --- /dev/null +++ b/fixtures/golden/heico-student-status/inputs/export-exclusions.json @@ -0,0 +1,6 @@ +[ + { + "student_id": "S-200", + "reason": "No active enrollment" + } +] diff --git a/fixtures/golden/heico-student-status/inputs/student-statuses.json b/fixtures/golden/heico-student-status/inputs/student-statuses.json new file mode 100644 index 0000000..2ff7877 --- /dev/null +++ b/fixtures/golden/heico-student-status/inputs/student-statuses.json @@ -0,0 +1,38 @@ +[ + { + "student_id": "S-100", + "first_name": " Ada ", + "last_name": "Lovelace", + "birth_date": "2000-12-10", + "program": "CS", + "status": "active", + "status_date": "2026-04-01" + }, + { + "student_id": "S-100", + "first_name": " Ada ", + "last_name": "Lovelace", + "birth_date": "2000-12-10", + "program": "CS", + "status": "enrolled", + "status_date": "2025-10-01" + }, + { + "student_id": "S-200", + "first_name": "Grace", + "last_name": "Hopper", + "birth_date": "1999-12-09", + "program": "IS", + "status": "withdrawn", + "status_date": "2026-03-15" + }, + { + "student_id": "S-300", + "first_name": "Katherine", + "last_name": "Johnson", + "birth_date": "2001-08-26", + "program": "MATH", + "status": "active", + "status_date": "2026-04-02" + } +] diff --git a/fixtures/golden/heico-student-status/manifest.json b/fixtures/golden/heico-student-status/manifest.json new file mode 100644 index 0000000..0e49d4d --- /dev/null +++ b/fixtures/golden/heico-student-status/manifest.json @@ -0,0 +1,7 @@ +{ + "id": "heico-student-status", + "title": "HEICO-style current student status export", + "user_story": "GovOPlaN/govoplan-dataflow#17", + "graph": "graph.json", + "expected_output": "expected-output.json" +} diff --git a/fixtures/golden/rele-booking-transform/expected-output.json b/fixtures/golden/rele-booking-transform/expected-output.json new file mode 100644 index 0000000..d5d2e14 --- /dev/null +++ b/fixtures/golden/rele-booking-transform/expected-output.json @@ -0,0 +1,20 @@ +[ + { + "period": "2026/01", + "booking_class": "personnel", + "booking_count": 2, + "net_amount": 70 + }, + { + "period": "2026/02", + "booking_class": "allocation", + "booking_count": 1, + "net_amount": 45 + }, + { + "period": "2026/02", + "booking_class": "other", + "booking_count": 1, + "net_amount": 20 + } +] diff --git a/fixtures/golden/rele-booking-transform/graph.json b/fixtures/golden/rele-booking-transform/graph.json new file mode 100644 index 0000000..c5ca58f --- /dev/null +++ b/fixtures/golden/rele-booking-transform/graph.json @@ -0,0 +1,104 @@ +{ + "schema_version": 1, + "nodes": [ + { + "id": "bookings", + "type": "source.inline", + "label": "Bookings", + "position": {"x": 40, "y": 120}, + "config": { + "source_name": "bookings", + "fixture": "bookings.json", + "rows": [] + } + }, + { + "id": "booking-calculations", + "type": "calculate", + "label": "Normalize and classify", + "position": {"x": 280, "y": 120}, + "config": { + "calculations": [ + { + "target_column": "cost_center", + "expression": "cast(trim(cost_text) as integer)", + "result_type": "integer" + }, + { + "target_column": "period", + "expression": "to_char(booking_date, 'YYYY/MM')", + "result_type": "string" + }, + { + "target_column": "booking_class", + "expression": "case when cost_center between 62000 and 65999 then 'personnel' when cost_center between 97000000 and 97999999 then 'allocation' else 'other' end", + "result_type": "string" + }, + { + "target_column": "signed_amount", + "expression": "case when entry_type = 'credit' then 0 - amount else amount end", + "result_type": "integer" + }, + { + "target_column": "record_key", + "expression": "lpad(cast(sequence as text), 6, '0')", + "result_type": "string" + } + ] + } + }, + { + "id": "required-fields", + "type": "quality.rules", + "label": "Required booking fields", + "position": {"x": 520, "y": 120}, + "config": { + "rules": [ + {"id": "cost-center", "column": "cost_center", "operator": "not_null"}, + {"id": "period", "column": "period", "operator": "not_null"}, + {"id": "amount", "column": "signed_amount", "operator": "not_null"} + ], + "action": "fail" + } + }, + { + "id": "booking-summary", + "type": "aggregate", + "label": "Monthly booking summary", + "position": {"x": 760, "y": 120}, + "config": { + "group_by": ["period", "booking_class"], + "aggregates": [ + {"function": "count", "column": "*", "alias": "booking_count"}, + {"function": "sum", "column": "signed_amount", "alias": "net_amount"} + ] + } + }, + { + "id": "summary-order", + "type": "sort", + "label": "Order summary", + "position": {"x": 1000, "y": 120}, + "config": { + "fields": [ + {"column": "period", "direction": "asc"}, + {"column": "booking_class", "direction": "asc"} + ] + } + }, + { + "id": "output", + "type": "output", + "label": "Booking summary", + "position": {"x": 1240, "y": 120}, + "config": {} + } + ], + "edges": [ + {"id": "e1", "source": "bookings", "target": "booking-calculations"}, + {"id": "e2", "source": "booking-calculations", "target": "required-fields"}, + {"id": "e3", "source": "required-fields", "target": "booking-summary"}, + {"id": "e4", "source": "booking-summary", "target": "summary-order"}, + {"id": "e5", "source": "summary-order", "target": "output"} + ] +} diff --git a/fixtures/golden/rele-booking-transform/inputs/bookings.json b/fixtures/golden/rele-booking-transform/inputs/bookings.json new file mode 100644 index 0000000..74e2300 --- /dev/null +++ b/fixtures/golden/rele-booking-transform/inputs/bookings.json @@ -0,0 +1,30 @@ +[ + { + "sequence": 1, + "booking_date": "2026-01-12", + "cost_text": " 62150 ", + "entry_type": "debit", + "amount": 100 + }, + { + "sequence": 2, + "booking_date": "2026-01-13", + "cost_text": "62210", + "entry_type": "credit", + "amount": 30 + }, + { + "sequence": 3, + "booking_date": "2026-02-02", + "cost_text": "97020500", + "entry_type": "debit", + "amount": 45 + }, + { + "sequence": 4, + "booking_date": "2026-02-03", + "cost_text": "81000", + "entry_type": "debit", + "amount": 20 + } +] diff --git a/fixtures/golden/rele-booking-transform/manifest.json b/fixtures/golden/rele-booking-transform/manifest.json new file mode 100644 index 0000000..fa7e070 --- /dev/null +++ b/fixtures/golden/rele-booking-transform/manifest.json @@ -0,0 +1,7 @@ +{ + "id": "rele-booking-transform", + "title": "RELE-style booking normalization and aggregation", + "user_story": "GovOPlaN/govoplan-dataflow#17", + "graph": "graph.json", + "expected_output": "expected-output.json" +} diff --git a/src/govoplan_dataflow/backend/executor.py b/src/govoplan_dataflow/backend/executor.py index 4c29ca2..dc16008 100644 --- a/src/govoplan_dataflow/backend/executor.py +++ b/src/govoplan_dataflow/backend/executor.py @@ -498,6 +498,16 @@ def _join_rows( for left_row in left_rows: key = _join_key(left_row, left_keys) matches = right_index.get(key, ()) if key is not None else () + if join_type == "semi": + if matches: + output.append(dict(left_row)) + _guard_intermediate_size(output, node_id=node_id) + continue + if join_type == "anti": + if not matches: + output.append(dict(left_row)) + _guard_intermediate_size(output, node_id=node_id) + continue if matches: for right_index_value, right_row in matches: matched_right.add(right_index_value) @@ -688,6 +698,41 @@ def _expression_rows( return output +def _calculation_rows( + rows: list[dict[str, Any]], + config: dict[str, Any], + *, + node_id: str, +) -> list[dict[str, Any]]: + calculations: list[tuple[str, Any]] = [] + try: + calculations = [ + ( + str(item["target_column"]), + parse_expression(str(item["expression"])), + ) + for item in config["calculations"] + ] + except (KeyError, TypeError, ValueError) as exc: + raise PipelineExecutionError( + f"Cannot prepare calculated columns: {exc}", + node_id=node_id, + ) from exc + output: list[dict[str, Any]] = [] + for row in rows: + result = dict(row) + for target, parsed in calculations: + try: + result[target] = evaluate_expression(parsed, result) + except (ArithmeticError, TypeError, ValueError) as exc: + raise PipelineExecutionError( + f"Cannot calculate {target!r}: {exc}", + node_id=node_id, + ) from exc + output.append(result) + return output + + def _convert_rows( rows: list[dict[str, Any]], config: dict[str, Any], @@ -1165,6 +1210,65 @@ def _sort_rows(rows: list[dict[str, Any]], config: dict[str, Any]) -> list[dict[ return result +def _rank_rows( + rows: list[dict[str, Any]], + config: dict[str, Any], +) -> list[dict[str, Any]]: + partition_by = [str(item) for item in config.get("partition_by", [])] + order_by = list(config["order_by"]) + partitions: dict[tuple[Any, ...], list[tuple[int, dict[str, Any]]]] = ( + defaultdict(list) + ) + for index, row in enumerate(rows): + key = tuple(_hashable(row.get(column)) for column in partition_by) + partitions[key].append((index, row)) + + ranks: dict[int, int] = {} + method = str(config.get("method", "row_number")) + for partition in partitions.values(): + ordered = list(partition) + for field_config in reversed(order_by): + column = str(field_config["column"]) + reverse = field_config.get("direction", "asc") == "desc" + concrete = [ + item for item in ordered if item[1].get(column) is not None + ] + nulls = [ + item for item in ordered if item[1].get(column) is None + ] + concrete.sort( + key=lambda item: _sortable_value(item[1][column]), + reverse=reverse, + ) + ordered = [*concrete, *nulls] + + previous_values: tuple[Any, ...] | None = None + current_rank = 0 + dense_rank = 0 + for position, (source_index, row) in enumerate(ordered, start=1): + values = tuple( + _hashable(row.get(str(field["column"]))) + for field in order_by + ) + if previous_values is None or values != previous_values: + current_rank = position + dense_rank += 1 + previous_values = values + ranks[source_index] = ( + position + if method == "row_number" + else dense_rank + if method == "dense_rank" + else current_rank + ) + + target = str(config["target_column"]) + return [ + {**row, target: ranks[index]} + for index, row in enumerate(rows) + ] + + def _sortable_value(value: Any) -> tuple[str, Any]: if isinstance(value, (int, float, Decimal, str)): return type(value).__name__, value @@ -1336,6 +1440,13 @@ def _register_executors() -> None: node_id=context.node.id, ) ), + "calculate": lambda context: OperatorExecutionResult( + rows=_calculation_rows( + context.input_rows, + context.node.config, + node_id=context.node.id, + ) + ), "convert": lambda context: OperatorExecutionResult( rows=_convert_rows( context.input_rows, @@ -1356,6 +1467,9 @@ def _register_executors() -> None: "sort": lambda context: OperatorExecutionResult( rows=_sort_rows(context.input_rows, context.node.config) ), + "window.rank": lambda context: OperatorExecutionResult( + rows=_rank_rows(context.input_rows, context.node.config) + ), "limit": lambda context: OperatorExecutionResult( rows=context.input_rows[: int(context.node.config["count"])] ), diff --git a/src/govoplan_dataflow/backend/expressions.py b/src/govoplan_dataflow/backend/expressions.py index 55aa778..f45d75b 100644 --- a/src/govoplan_dataflow/backend/expressions.py +++ b/src/govoplan_dataflow/backend/expressions.py @@ -1,7 +1,10 @@ from __future__ import annotations +import calendar import operator +import re from dataclasses import dataclass +from datetime import date, datetime, timedelta from decimal import Decimal from typing import Any, Callable, Literal @@ -36,6 +39,12 @@ class ParsedExpression: return self.expression.sql(dialect="duckdb") +@dataclass(frozen=True, slots=True) +class _IntervalValue: + amount: int + unit: str + + _LEAF_TYPES = ( exp.Column, exp.Identifier, @@ -43,6 +52,8 @@ _LEAF_TYPES = ( exp.Null, exp.Boolean, exp.DataType, + exp.Interval, + exp.Var, ) _BINARY_TYPES = ( exp.Add, @@ -50,6 +61,7 @@ _BINARY_TYPES = ( exp.Mul, exp.Div, exp.Mod, + exp.DPipe, exp.EQ, exp.NEQ, exp.GT, @@ -60,6 +72,14 @@ _BINARY_TYPES = ( exp.Or, exp.Is, ) +_PREDICATE_TYPES = ( + exp.Between, + exp.ILike, + exp.In, + exp.Like, + exp.RegexpFullMatch, + exp.RegexpLike, +) _UNARY_TYPES = (exp.Not, exp.Neg, exp.Paren) _FUNCTION_TYPES = ( exp.Lower, @@ -73,11 +93,30 @@ _FUNCTION_TYPES = ( exp.Replace, exp.Substring, exp.Cast, + exp.ConcatWs, + exp.DateAdd, + exp.DateDiff, + exp.DateSub, + exp.Day, + exp.Extract, + exp.Greatest, + exp.Least, + exp.Month, + exp.Nullif, + exp.Pad, + exp.RegexpReplace, + exp.SplitPart, + exp.TimeToStr, + exp.TimestampTrunc, + exp.ToChar, + exp.Translate, + exp.Year, ) _CONTROL_TYPES = (exp.Case, exp.If) _ALLOWED_TYPES = ( *_LEAF_TYPES, *_BINARY_TYPES, + *_PREDICATE_TYPES, *_UNARY_TYPES, *_FUNCTION_TYPES, *_CONTROL_TYPES, @@ -182,6 +221,81 @@ def _evaluate_is(expression: exp.Expression, row: dict[str, Any]) -> bool: return left is right if right is None else left == right +def _evaluate_like(expression: exp.Expression, row: dict[str, Any]) -> bool: + value = _evaluate(expression.this, row) + pattern = _evaluate(expression.expression, row) + if value is None or pattern is None: + return False + flags = re.IGNORECASE if isinstance(expression, exp.ILike) else 0 + return ( + re.fullmatch( + _sql_like_pattern(str(pattern)), + str(value), + flags=flags, + ) + is not None + ) + + +def _sql_like_pattern(pattern: str) -> str: + pieces: list[str] = [] + escaped = False + for character in pattern: + if escaped: + pieces.append(re.escape(character)) + escaped = False + elif character == "\\": + escaped = True + elif character == "%": + pieces.append(".*") + elif character == "_": + pieces.append(".") + else: + pieces.append(re.escape(character)) + if escaped: + pieces.append(re.escape("\\")) + return "".join(pieces) + + +def _evaluate_in(expression: exp.Expression, row: dict[str, Any]) -> bool: + value = _hashable_expression_value(_evaluate(expression.this, row)) + return value in { + _hashable_expression_value(_evaluate(item, row)) + for item in expression.expressions + } + + +def _hashable_expression_value(value: Any) -> Any: + if isinstance(value, (dict, list, set, tuple)): + return repr(value) + return value + + +def _evaluate_between(expression: exp.Expression, row: dict[str, Any]) -> bool: + value = _evaluate(expression.this, row) + low = _evaluate(expression.args["low"], row) + high = _evaluate(expression.args["high"], row) + if value is None or low is None or high is None: + return False + return low <= value <= high + + +def _evaluate_regexp_match( + expression: exp.Expression, + row: dict[str, Any], +) -> bool: + value = _evaluate(expression.this, row) + pattern = _evaluate(expression.expression, row) + if value is None or pattern is None: + return False + operation = ( + re.fullmatch + if isinstance(expression, exp.RegexpFullMatch) + else re.search + ) + return operation(str(pattern), str(value)) is not None + + def _evaluate_binary( expression: exp.Expression, row: dict[str, Any], @@ -197,6 +311,16 @@ def _evaluate_binary( def _evaluate_arithmetic(expression: exp.Expression, row: dict[str, Any]) -> Any: + if isinstance(expression, (exp.Add, exp.Sub)): + left = _evaluate(expression.this, row) + right = _evaluate(expression.expression, row) + if isinstance(right, _IntervalValue): + amount = -right.amount if isinstance(expression, exp.Sub) else right.amount + return _add_interval(left, amount, right.unit) + if isinstance(left, _IntervalValue): + if isinstance(expression, exp.Sub): + raise ValueError("An interval cannot be subtracted from an interval.") + return _add_interval(right, left.amount, left.unit) return _evaluate_binary( expression, row, @@ -232,6 +356,27 @@ def _evaluate_string_function( ) +def _evaluate_trim( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + if value is None: + return None + characters_expression = expression.args.get("expression") + characters = ( + str(_evaluate(characters_expression, row)) + if characters_expression is not None + else None + ) + position = str(expression.args.get("position") or "BOTH").upper() + if position == "LEADING": + return str(value).lstrip(characters) + if position == "TRAILING": + return str(value).rstrip(characters) + return str(value).strip(characters) + + def _evaluate_length(expression: exp.Expression, row: dict[str, Any]) -> int | None: value = _evaluate_child(expression, row) return None if value is None else len(value) @@ -286,6 +431,426 @@ def _evaluate_substring(expression: exp.Expression, row: dict[str, Any]) -> Any: return str(value)[start : start + int(_evaluate(length, row))] +def _evaluate_pad(expression: exp.Expression, row: dict[str, Any]) -> str | None: + value = _evaluate_child(expression, row) + if value is None: + return None + target_length = int(_evaluate(expression.expression, row)) + if target_length < 0: + raise ValueError("Padding length cannot be negative.") + source = str(value) + if len(source) >= target_length: + return source[:target_length] + fill_expression = expression.args.get("fill_pattern") + fill = ( + str(_evaluate(fill_expression, row)) + if fill_expression is not None + else " " + ) + if not fill: + raise ValueError("Padding fill text cannot be empty.") + padding_length = target_length - len(source) + padding = (fill * ((padding_length // len(fill)) + 1))[:padding_length] + return ( + f"{padding}{source}" + if expression.args.get("is_left") + else f"{source}{padding}" + ) + + +def _evaluate_split_part( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + delimiter = _evaluate(expression.args["delimiter"], row) + part_index = int(_evaluate(expression.args["part_index"], row)) + if value is None or delimiter is None: + return None + if not delimiter: + raise ValueError("SPLIT_PART delimiter cannot be empty.") + if part_index == 0: + raise ValueError("SPLIT_PART index is one-based.") + parts = str(value).split(str(delimiter)) + index = part_index - 1 if part_index > 0 else part_index + return parts[index] if -len(parts) <= index < len(parts) else "" + + +def _evaluate_translate( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + source = _evaluate(expression.args["from_"], row) + target = _evaluate(expression.args["to"], row) + if value is None or source is None or target is None: + return None + target_text = str(target) + translation = { + ord(character): ( + target_text[index] + if index < len(target_text) + else None + ) + for index, character in enumerate(str(source)) + } + return str(value).translate(translation) + + +def _evaluate_regexp_replace( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + pattern = _evaluate(expression.expression, row) + replacement = _evaluate(expression.args["replacement"], row) + if value is None or pattern is None or replacement is None: + return None + modifiers_expression = expression.args.get("modifiers") + modifiers = ( + str(_evaluate(modifiers_expression, row)) + if modifiers_expression is not None + else "" + ) + flags = re.IGNORECASE if "i" in modifiers else 0 + count = 0 if "g" in modifiers else 1 + return re.sub( + str(pattern), + str(replacement), + str(value), + count=count, + flags=flags, + ) + + +def _evaluate_concat_ws( + expression: exp.Expression, + row: dict[str, Any], +) -> str: + arguments = list(expression.expressions) + separator_value = _evaluate(arguments[0], row) if arguments else None + separator = "" if separator_value is None else str(separator_value) + return separator.join( + str(value) + for argument in arguments[1:] + if (value := _evaluate(argument, row)) is not None + ) + + +def _evaluate_pipe( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + left = _evaluate(expression.this, row) + right = _evaluate(expression.expression, row) + return None if left is None or right is None else f"{left}{right}" + + +def _evaluate_nullif( + expression: exp.Expression, + row: dict[str, Any], +) -> Any: + value = _evaluate_child(expression, row) + comparison = _evaluate(expression.expression, row) + return None if value == comparison else value + + +def _evaluate_extreme( + expression: exp.Expression, + row: dict[str, Any], +) -> Any: + values = [ + value + for argument in (expression.this, *expression.expressions) + if (value := _evaluate(argument, row)) is not None + ] + if not values: + return None + return ( + max(values) + if isinstance(expression, exp.Greatest) + else min(values) + ) + + +def _evaluate_time_to_str( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + format_expression = expression.args.get("format") + format_text = ( + _evaluate(format_expression, row) + if format_expression is not None + else None + ) + if value is None or format_text is None: + return None + return _format_temporal(value, str(format_text), postgres=False) + + +def _evaluate_to_char( + expression: exp.Expression, + row: dict[str, Any], +) -> str | None: + value = _evaluate_child(expression, row) + format_expression = expression.args.get("format") + format_text = ( + _evaluate(format_expression, row) + if format_expression is not None + else None + ) + if value is None or format_text is None: + return None + if not isinstance(value, (date, datetime)) and not _looks_temporal(value): + raise ValueError( + "TO_CHAR currently supports date and date-time values only." + ) + return _format_temporal(value, str(format_text), postgres=True) + + +def _format_temporal( + value: Any, + format_text: str, + *, + postgres: bool, +) -> str: + if not postgres: + return _temporal_value(value).strftime(format_text) + translated = format_text + for source, target in ( + ("TMMonth", "%B"), + ("TMMon", "%b"), + ("TMYYYY", "%Y"), + ("YYYY", "%Y"), + ("YYY", "%Y"), + ("YY", "%y"), + ("MM", "%m"), + ("DD", "%d"), + ("HH24", "%H"), + ("MI", "%M"), + ("SS", "%S"), + ): + translated = translated.replace(source, target) + return _temporal_value(value).strftime(translated) + + +def _looks_temporal(value: Any) -> bool: + text = str(value or "").strip() + if not text: + return False + try: + _temporal_value(text) + except ValueError: + return False + return True + + + + +def _evaluate_extract( + expression: exp.Expression, + row: dict[str, Any], +) -> int: + value = _temporal_value(_evaluate(expression.expression, row)) + unit = _date_unit(expression.this) + return _temporal_part(value, unit) + + +def _evaluate_temporal_part( + expression: exp.Expression, + row: dict[str, Any], +) -> int: + value = _temporal_value(_evaluate_child(expression, row)) + unit = { + exp.Year: "year", + exp.Month: "month", + exp.Day: "day", + }[type(expression)] + return _temporal_part(value, unit) + + +def _temporal_part(value: date | datetime, unit: str) -> int: + mapping = { + "year": value.year, + "quarter": ((value.month - 1) // 3) + 1, + "month": value.month, + "week": int(value.strftime("%V")), + "day": value.day, + "dow": value.weekday(), + "doy": int(value.strftime("%j")), + "hour": value.hour if isinstance(value, datetime) else 0, + "minute": value.minute if isinstance(value, datetime) else 0, + "second": value.second if isinstance(value, datetime) else 0, + } + if unit not in mapping: + raise ValueError(f"Unsupported date part {unit!r}.") + return mapping[unit] + + +def _evaluate_timestamp_trunc( + expression: exp.Expression, + row: dict[str, Any], +) -> date | datetime: + value = _temporal_value(_evaluate_child(expression, row)) + unit = _date_unit(expression.args["unit"]) + if unit == "year": + return value.replace(month=1, day=1, **_time_reset(value)) + if unit == "quarter": + month = ((value.month - 1) // 3) * 3 + 1 + return value.replace(month=month, day=1, **_time_reset(value)) + if unit == "month": + return value.replace(day=1, **_time_reset(value)) + if unit == "week": + result = value - timedelta(days=value.weekday()) + return result.replace(**_time_reset(result)) + if unit == "day": + return value.replace(**_time_reset(value)) + if not isinstance(value, datetime): + raise ValueError(f"Cannot truncate a date to {unit}.") + if unit == "hour": + return value.replace(minute=0, second=0, microsecond=0) + if unit == "minute": + return value.replace(second=0, microsecond=0) + if unit == "second": + return value.replace(microsecond=0) + raise ValueError(f"Unsupported date truncation unit {unit!r}.") + + +def _time_reset(value: date | datetime) -> dict[str, int]: + return ( + {"hour": 0, "minute": 0, "second": 0, "microsecond": 0} + if isinstance(value, datetime) + else {} + ) + + +def _evaluate_date_add( + expression: exp.Expression, + row: dict[str, Any], +) -> date | datetime: + value = _evaluate_child(expression, row) + amount, unit = _interval_value(expression.expression, row) + if isinstance(expression, (exp.DateSub, exp.Sub)): + amount = -amount + return _add_interval(value, amount, unit) + + +def _add_interval( + value: Any, + amount: int, + unit: str, +) -> date | datetime: + temporal = _temporal_value(value) + if unit in {"year", "quarter", "month"}: + months = amount * {"year": 12, "quarter": 3, "month": 1}[unit] + return _add_months(temporal, months) + seconds = amount * { + "week": 7 * 24 * 60 * 60, + "day": 24 * 60 * 60, + "hour": 60 * 60, + "minute": 60, + "second": 1, + }.get(unit, 0) + if unit not in {"week", "day", "hour", "minute", "second"}: + raise ValueError(f"Unsupported interval unit {unit!r}.") + return temporal + timedelta(seconds=seconds) + + +def _evaluate_date_diff( + expression: exp.Expression, + row: dict[str, Any], +) -> int: + end = _temporal_value(_evaluate_child(expression, row)) + start = _temporal_value(_evaluate(expression.expression, row)) + unit = _date_unit(expression.args["unit"]) + if unit == "year": + return end.year - start.year + if unit in {"quarter", "month"}: + months = (end.year - start.year) * 12 + end.month - start.month + return months // 3 if unit == "quarter" else months + delta = datetime.combine(end, datetime.min.time()) - datetime.combine( + start, + datetime.min.time(), + ) if not isinstance(end, datetime) and not isinstance(start, datetime) else ( + _as_datetime(end) - _as_datetime(start) + ) + divisor = { + "week": 7 * 24 * 60 * 60, + "day": 24 * 60 * 60, + "hour": 60 * 60, + "minute": 60, + "second": 1, + }.get(unit) + if divisor is None: + raise ValueError(f"Unsupported date difference unit {unit!r}.") + return int(delta.total_seconds() / divisor) + + +def _interval_value( + expression: exp.Expression, + row: dict[str, Any], +) -> tuple[int, str]: + if not isinstance(expression, exp.Interval): + raise ValueError("Date arithmetic requires an INTERVAL.") + interval = _evaluate_interval(expression, row) + return interval.amount, interval.unit + + +def _evaluate_interval( + expression: exp.Expression, + row: dict[str, Any], +) -> _IntervalValue: + return _IntervalValue( + amount=int(_evaluate(expression.this, row)), + unit=_date_unit(expression.args["unit"]), + ) + + +def _date_unit(expression: exp.Expression) -> str: + unit = str(expression.this).casefold() + aliases = { + "years": "year", + "quarters": "quarter", + "months": "month", + "weeks": "week", + "days": "day", + "hours": "hour", + "minutes": "minute", + "seconds": "second", + } + return aliases.get(unit, unit) + + +def _temporal_value(value: Any) -> date | datetime: + if isinstance(value, datetime): + return value + if isinstance(value, date): + return value + text = str(value or "").strip() + if not text: + raise ValueError("A date or date-time value is required.") + if "T" not in text and " " not in text: + return date.fromisoformat(text) + return datetime.fromisoformat(text.replace("Z", "+00:00")) + + +def _as_datetime(value: date | datetime) -> datetime: + return ( + value + if isinstance(value, datetime) + else datetime.combine(value, datetime.min.time()) + ) + + +def _add_months(value: date | datetime, months: int) -> date | datetime: + month_index = value.year * 12 + value.month - 1 + months + year, zero_based_month = divmod(month_index, 12) + month = zero_based_month + 1 + day = min(value.day, calendar.monthrange(year, month)[1]) + return value.replace(year=year, month=month, day=day) + + def _evaluate_cast(expression: exp.Expression, row: dict[str, Any]) -> Any: return convert_value( _evaluate_child(expression, row), @@ -350,7 +915,6 @@ _COMPARISON_OPERATIONS: dict[ _STRING_OPERATIONS: dict[type[exp.Expression], Callable[[str], str]] = { exp.Lower: str.lower, exp.Upper: str.upper, - exp.Trim: str.strip, } _EVALUATORS: dict[ type[exp.Expression], @@ -361,11 +925,18 @@ _EVALUATORS: dict[ exp.Null: lambda _expression, _row: None, exp.Boolean: lambda expression, _row: bool(expression.this), exp.Literal: lambda expression, _row: _literal(expression), # type: ignore[arg-type] + exp.Interval: _evaluate_interval, exp.Neg: _evaluate_negation, exp.Not: _evaluate_boolean_not, exp.And: _evaluate_boolean_binary, exp.Or: _evaluate_boolean_binary, exp.Is: _evaluate_is, + exp.Like: _evaluate_like, + exp.ILike: _evaluate_like, + exp.In: _evaluate_in, + exp.Between: _evaluate_between, + exp.RegexpLike: _evaluate_regexp_match, + exp.RegexpFullMatch: _evaluate_regexp_match, **{ expression_type: _evaluate_arithmetic for expression_type in _ARITHMETIC_OPERATIONS @@ -379,12 +950,32 @@ _EVALUATORS: dict[ for expression_type in _STRING_OPERATIONS }, exp.Length: _evaluate_length, + exp.Trim: _evaluate_trim, exp.Abs: _evaluate_abs, exp.Round: _evaluate_round, exp.Coalesce: _evaluate_coalesce, exp.Concat: _evaluate_concat, + exp.ConcatWs: _evaluate_concat_ws, + exp.DPipe: _evaluate_pipe, + exp.Greatest: _evaluate_extreme, + exp.Least: _evaluate_extreme, + exp.Nullif: _evaluate_nullif, + exp.Pad: _evaluate_pad, exp.Replace: _evaluate_replace, + exp.RegexpReplace: _evaluate_regexp_replace, + exp.SplitPart: _evaluate_split_part, exp.Substring: _evaluate_substring, + exp.TimeToStr: _evaluate_time_to_str, + exp.ToChar: _evaluate_to_char, + exp.Translate: _evaluate_translate, + exp.Extract: _evaluate_extract, + exp.Year: _evaluate_temporal_part, + exp.Month: _evaluate_temporal_part, + exp.Day: _evaluate_temporal_part, + exp.TimestampTrunc: _evaluate_timestamp_trunc, + exp.DateAdd: _evaluate_date_add, + exp.DateSub: _evaluate_date_add, + exp.DateDiff: _evaluate_date_diff, exp.Cast: _evaluate_cast, exp.Case: _evaluate_case, exp.If: _evaluate_if, @@ -462,6 +1053,11 @@ def _infer_numeric( expression: exp.Expression, schema: dict[str, ExpressionDataType], ) -> ExpressionDataType: + if ( + isinstance(expression, (exp.Add, exp.Sub)) + and isinstance(expression.expression, exp.Interval) + ): + return _infer(expression.this, schema) child_types = { _infer(item, schema) for item in expression.iter_expressions() @@ -485,6 +1081,18 @@ def _infer_coalesce( ) +def _infer_extreme( + expression: exp.Expression, + schema: dict[str, ExpressionDataType], +) -> ExpressionDataType: + return _common_expression_type( + [ + _infer(item, schema) + for item in (expression.this, *expression.expressions) + ] + ) + + def _infer_case( expression: exp.Expression, schema: dict[str, ExpressionDataType], @@ -553,6 +1161,12 @@ _TYPE_INFERERS: dict[ exp.And, exp.Or, exp.Is, + exp.Like, + exp.ILike, + exp.In, + exp.Between, + exp.RegexpLike, + exp.RegexpFullMatch, exp.Not, ) }, @@ -564,8 +1178,16 @@ _TYPE_INFERERS: dict[ exp.Upper, exp.Trim, exp.Concat, + exp.ConcatWs, + exp.DPipe, + exp.Pad, exp.Replace, + exp.RegexpReplace, + exp.SplitPart, exp.Substring, + exp.TimeToStr, + exp.ToChar, + exp.Translate, ) }, exp.Cast: _infer_cast, @@ -582,6 +1204,17 @@ _TYPE_INFERERS: dict[ ) }, exp.Coalesce: _infer_coalesce, + exp.Greatest: _infer_extreme, + exp.Least: _infer_extreme, + exp.Nullif: _infer_child, + exp.Extract: lambda _expression, _schema: "integer", + exp.Year: lambda _expression, _schema: "integer", + exp.Month: lambda _expression, _schema: "integer", + exp.Day: lambda _expression, _schema: "integer", + exp.DateDiff: lambda _expression, _schema: "integer", + exp.DateAdd: _infer_child, + exp.DateSub: _infer_child, + exp.TimestampTrunc: _infer_child, exp.Case: _infer_case, exp.If: _infer_if, exp.Paren: _infer_child, diff --git a/src/govoplan_dataflow/backend/graph.py b/src/govoplan_dataflow/backend/graph.py index 6bf6417..6ae27fe 100644 --- a/src/govoplan_dataflow/backend/graph.py +++ b/src/govoplan_dataflow/backend/graph.py @@ -47,7 +47,8 @@ DERIVE_OPERATIONS = frozenset( "divide", } ) -JOIN_TYPES = frozenset({"inner", "left", "right", "full"}) +JOIN_TYPES = frozenset({"inner", "left", "right", "full", "semi", "anti"}) +RANK_METHODS = frozenset({"row_number", "rank", "dense_rank"}) DATA_TYPES = frozenset( {"string", "integer", "number", "boolean", "date", "datetime"} ) @@ -901,6 +902,65 @@ def _validate_expression(node: GraphNode) -> list[DataflowDiagnostic]: return diagnostics +def _validate_calculate(node: GraphNode) -> list[DataflowDiagnostic]: + calculations = node.config.get("calculations") + if not isinstance(calculations, list) or not 1 <= len(calculations) <= 100: + return [ + _node_field_error( + node, + "calculate.required", + "Add between one and 100 calculated columns.", + "calculations", + ) + ] + diagnostics: list[DataflowDiagnostic] = [] + targets: list[str] = [] + for item in calculations: + if not isinstance(item, dict): + diagnostics.append( + _node_field_error( + node, + "calculate.item", + "Every calculation needs a target column and expression.", + "calculations", + ) + ) + continue + target = str(item.get("target_column") or "").strip() + expression = str(item.get("expression") or "").strip() + result_type = str(item.get("result_type") or "unknown") + if not target or not expression: + diagnostics.append( + _node_field_error( + node, + "calculate.item", + "Every calculation needs a target column and expression.", + "calculations", + ) + ) + if target: + targets.append(target) + if result_type not in {"unknown", *DATA_TYPES}: + diagnostics.append( + _node_field_error( + node, + "calculate.result_type", + "Choose a supported calculation result type.", + "calculations", + ) + ) + if len(targets) != len(set(targets)): + diagnostics.append( + _node_field_error( + node, + "calculate.duplicate_target", + "Calculated column names must be unique within the block.", + "calculations", + ) + ) + return diagnostics + + def _validate_convert(node: GraphNode) -> list[DataflowDiagnostic]: diagnostics = _validate_source_target_columns(node, "convert") if node.config.get("target_type") not in DATA_TYPES: @@ -988,6 +1048,56 @@ def _validate_sort(node: GraphNode) -> list[DataflowDiagnostic]: ] +def _validate_rank(node: GraphNode) -> list[DataflowDiagnostic]: + diagnostics: list[DataflowDiagnostic] = [] + if node.config.get("method", "row_number") not in RANK_METHODS: + diagnostics.append( + _node_field_error( + node, + "rank.method", + "Choose row number, rank, or dense rank.", + "method", + ) + ) + if not _non_empty_text(node.config.get("target_column")): + diagnostics.append( + _node_field_error( + node, + "rank.target_column", + "Choose an output column.", + "target_column", + ) + ) + partition_by = node.config.get("partition_by", []) + if not isinstance(partition_by, list) or any( + not _non_empty_text(item) for item in partition_by + ): + diagnostics.append( + _node_field_error( + node, + "rank.partition_by", + "Partition columns must be named.", + "partition_by", + ) + ) + order_by = node.config.get("order_by") + if not isinstance(order_by, list) or not order_by or not all( + isinstance(item, dict) + and _non_empty_text(item.get("column")) + and item.get("direction", "asc") in {"asc", "desc"} + for item in order_by + ): + diagnostics.append( + _node_field_error( + node, + "rank.order_by", + "Add at least one ordered column with a valid direction.", + "order_by", + ) + ) + return diagnostics + + def _validate_limit(node: GraphNode) -> list[DataflowDiagnostic]: count = node.config.get("count") if ( @@ -1296,10 +1406,12 @@ def _register_config_validators() -> None: "select": _validate_select, "derive": _validate_derive, "expression": _validate_expression, + "calculate": _validate_calculate, "convert": _validate_convert, "replace": _validate_replace, "aggregate": _validate_aggregate, "sort": _validate_sort, + "window.rank": _validate_rank, "limit": _validate_limit, "quality.rules": _validate_quality, "reconcile.compare": _validate_reconcile, @@ -1323,6 +1435,7 @@ __all__ = [ "DERIVE_OPERATIONS", "FILTER_OPERATORS", "JOIN_TYPES", + "RANK_METHODS", "SUPPORTED_NODE_TYPES", "canonical_graph_payload", "definition_hash", diff --git a/src/govoplan_dataflow/backend/ir.py b/src/govoplan_dataflow/backend/ir.py index fcbc7d4..eefe293 100644 --- a/src/govoplan_dataflow/backend/ir.py +++ b/src/govoplan_dataflow/backend/ir.py @@ -279,26 +279,43 @@ def _node_expressions( node: GraphNode, output_state: SchemaState, ) -> tuple[IrExpression, ...]: + if node.type == "calculate": + sources = [ + str(item.get("expression") or "") + for item in node.config.get("calculations", []) + if isinstance(item, dict) and item.get("expression") + ] + return tuple( + item + for source in sources + if (item := _ir_expression(source, output_state)) is not None + ) source = node.config.get("expression") if node.type not in {"expression", "filter.expression"} or not source: return () + expression = _ir_expression(str(source), output_state) + return (expression,) if expression is not None else () + + +def _ir_expression( + source: str, + output_state: SchemaState, +) -> IrExpression | None: try: parsed = parse_expression(str(source)) except ExpressionError: - return () + return None inferred = infer_expression_type(parsed, output_state.types) - return ( - IrExpression( - source=parsed.source, - columns=parsed.columns, - result_type=_data_type(inferred), - semantic_hash=_hash_payload( - { - "dialect": "duckdb", - "source": parsed.sql(), - "result_type": inferred, - } - ), + return IrExpression( + source=parsed.source, + columns=parsed.columns, + result_type=_data_type(inferred), + semantic_hash=_hash_payload( + { + "dialect": "duckdb", + "source": parsed.sql(), + "result_type": inferred, + } ), ) diff --git a/src/govoplan_dataflow/backend/node_library.py b/src/govoplan_dataflow/backend/node_library.py index 0a366e3..9c4148b 100644 --- a/src/govoplan_dataflow/backend/node_library.py +++ b/src/govoplan_dataflow/backend/node_library.py @@ -117,6 +117,8 @@ _NODE_TYPES = ( ("left", "All left rows"), ("right", "All right rows"), ("full", "All rows"), + ("semi", "Left rows with a match"), + ("anti", "Left rows without a match"), ), ), NodeConfigField(id="left_keys", label="Left keys", kind="column_list", required=True), @@ -289,6 +291,35 @@ _NODE_TYPES = ( }, sql_support="partial", ), + NodeTypeDefinition( + type="calculate", + category="transform", + label="Calculate columns", + description=( + "Create or replace several columns using ordered safe typed " + "expressions." + ), + icon="calculator", + input_ports=(NodePortDefinition(id="input", label="Input"),), + config_fields=( + NodeConfigField( + id="calculations", + label="Calculations", + kind="calculations", + required=True, + ), + ), + default_config={ + "calculations": [ + { + "target_column": "", + "expression": "", + "result_type": "unknown", + } + ] + }, + sql_support="partial", + ), NodeTypeDefinition( type="convert", category="transform", @@ -386,6 +417,53 @@ _NODE_TYPES = ( config_fields=(NodeConfigField(id="fields", label="Sort fields", kind="sort_fields", required=True),), default_config={"fields": [{"column": "", "direction": "asc"}]}, ), + NodeTypeDefinition( + type="window.rank", + category="transform", + label="Rank rows", + description=( + "Number or rank rows within optional partitions using a stable " + "ordering." + ), + icon="list-ordered", + input_ports=(NodePortDefinition(id="input", label="Input"),), + config_fields=( + NodeConfigField( + id="method", + label="Method", + kind="select", + options=( + ("row_number", "Row number"), + ("rank", "Rank with gaps"), + ("dense_rank", "Dense rank"), + ), + ), + NodeConfigField( + id="target_column", + label="Output column", + kind="text", + required=True, + ), + NodeConfigField( + id="partition_by", + label="Partition by", + kind="column_list", + ), + NodeConfigField( + id="order_by", + label="Order by", + kind="sort_fields", + required=True, + ), + ), + default_config={ + "method": "row_number", + "target_column": "row_number", + "partition_by": [], + "order_by": [{"column": "", "direction": "asc"}], + }, + sql_support="partial", + ), NodeTypeDefinition( type="limit", category="transform", diff --git a/src/govoplan_dataflow/backend/schema_validation.py b/src/govoplan_dataflow/backend/schema_validation.py index 9c45855..f77836e 100644 --- a/src/govoplan_dataflow/backend/schema_validation.py +++ b/src/govoplan_dataflow/backend/schema_validation.py @@ -185,6 +185,8 @@ def _join(context: SchemaPropagationContext) -> SchemaPropagationResult: field_name="right_keys", ), ] + if node.config.get("join_type", "inner") in {"semi", "anti"}: + return SchemaPropagationResult(left_state, tuple(diagnostics)) prefix = str(node.config.get("right_prefix", "right_")) prefixed_right = { f"{prefix}{column}" @@ -377,6 +379,66 @@ def _expression(context: SchemaPropagationContext) -> SchemaPropagationResult: return SchemaPropagationResult(state, tuple(diagnostics)) +def _calculate(context: SchemaPropagationContext) -> SchemaPropagationResult: + node = context.node + state = context.input_state + diagnostics: list[DataflowDiagnostic] = [] + for item in _mapping_items(node.config.get("calculations")): + source = str(item.get("expression") or "") + target = str(item.get("target_column") or "") + try: + parsed = parse_expression(source) + except ExpressionError as exc: + diagnostics.append( + _error( + "expression.invalid", + str(exc), + node_id=node.id, + field="calculations", + ) + ) + continue + diagnostics.extend( + _unknown_columns( + node, + state, + list(parsed.columns), + field_name="calculations", + ) + ) + inferred = infer_expression_type( + parsed, + { + name: state.type_of(name) # type: ignore[dict-item] + for name in state.columns + }, + ) + expected = str(item.get("result_type") or "unknown") + if expected != "unknown" and inferred not in { + "unknown", + "null", + expected, + }: + diagnostics.append( + _warning( + "calculate.type_mismatch", + ( + f"Calculation for {target!r} infers {inferred}, " + f"not {expected}." + ), + node_id=node.id, + field="calculations", + ) + ) + if target: + state = _with_column( + state, + target, + expected if expected != "unknown" else inferred, + ) + return SchemaPropagationResult(state, tuple(diagnostics)) + + def _convert_or_replace( context: SchemaPropagationContext, ) -> SchemaPropagationResult: @@ -463,6 +525,31 @@ def _sort(context: SchemaPropagationContext) -> SchemaPropagationResult: ) +def _rank(context: SchemaPropagationContext) -> SchemaPropagationResult: + node = context.node + columns = [ + *_text_items(node.config.get("partition_by")), + *[ + str(item.get("column")) + for item in _mapping_items(node.config.get("order_by")) + if item.get("column") + ], + ] + diagnostics = _unknown_columns( + node, + context.input_state, + columns, + field_name="order_by", + ) + target = str(node.config.get("target_column") or "") + state = ( + _with_column(context.input_state, target, "integer") + if target + else context.input_state + ) + return SchemaPropagationResult(state, tuple(diagnostics)) + + def _quality(context: SchemaPropagationContext) -> SchemaPropagationResult: rules = _mapping_items(context.node.config.get("rules")) diagnostics = _unknown_columns( @@ -874,10 +961,12 @@ def register_schema_propagators() -> None: "select": _select, "derive": _derive, "expression": _expression, + "calculate": _calculate, "convert": _convert_or_replace, "replace": _convert_or_replace, "aggregate": _aggregate, "sort": _sort, + "window.rank": _rank, "limit": _identity, "quality.rules": _quality, "reconcile.compare": _reconcile, diff --git a/src/govoplan_dataflow/backend/sql_compiler.py b/src/govoplan_dataflow/backend/sql_compiler.py index a9a350c..31a0fa0 100644 --- a/src/govoplan_dataflow/backend/sql_compiler.py +++ b/src/govoplan_dataflow/backend/sql_compiler.py @@ -1128,11 +1128,37 @@ def _join_config( ) -> dict[str, Any]: kind = str(join.args.get("kind") or "").casefold() side = str(join.args.get("side") or "").casefold() - if kind not in {"", "inner", "outer"} or side not in {"", "left", "right", "full"}: + if kind not in {"", "inner", "outer", "semi", "anti"} or side not in { + "", + "left", + "right", + "full", + }: raise SqlCompilationError( - [_sql_error("sql.join_type", "JOIN supports INNER, LEFT, RIGHT, or FULL joins only.")] + [ + _sql_error( + "sql.join_type", + ( + "JOIN supports INNER, LEFT, RIGHT, FULL, SEMI, " + "or ANTI joins only." + ), + ) + ] ) - join_type = side or ("inner" if kind in {"", "inner"} else "") + if kind in {"semi", "anti"} and side: + raise SqlCompilationError( + [ + _sql_error( + "sql.join_type", + "SEMI and ANTI joins cannot use a side qualifier.", + ) + ] + ) + join_type = ( + kind + if kind in {"semi", "anti"} + else side or ("inner" if kind in {"", "inner"} else "") + ) if not join_type: raise SqlCompilationError( [_sql_error("sql.join_type", "OUTER JOIN requires LEFT, RIGHT, or FULL.")] @@ -1471,6 +1497,41 @@ def _render_expression( state.selected = True +def _render_calculate( + node: GraphNode, + state: _SqlRenderState, + column_expression: Callable[[str], exp.Column], +) -> None: + _require_projection_slot(node, state) + calculated_targets: set[str] = set() + expressions: list[exp.Expression] = [exp.Star()] + for item in node.config["calculations"]: + parsed = parse_expression(str(item["expression"])) + dependencies = set(parsed.columns) & calculated_targets + if dependencies: + raise SqlCompilationError( + [ + _node_sql_error( + node.id, + "sql.sequential_calculation", + ( + "SQL view cannot render a calculated column that " + "depends on an earlier calculation in the same block." + ), + ) + ] + ) + expression = _qualified_expression( + str(item["expression"]), + column_expression=column_expression, + ) + target = str(item["target_column"]) + expressions.append(expression.as_(target)) + calculated_targets.add(target) + state.select_expressions = expressions + state.selected = True + + def _render_derive( node: GraphNode, state: _SqlRenderState, @@ -1600,6 +1661,41 @@ def _render_sort( ] +def _render_rank( + node: GraphNode, + state: _SqlRenderState, + column_expression: Callable[[str], exp.Column], +) -> None: + _require_projection_slot(node, state) + function = { + "row_number": exp.RowNumber, + "rank": exp.Rank, + "dense_rank": exp.DenseRank, + }[str(node.config.get("method", "row_number"))]() + window = exp.Window( + this=function, + partition_by=[ + column_expression(str(column)) + for column in node.config.get("partition_by", []) + ], + order=exp.Order( + expressions=[ + exp.Ordered( + this=column_expression(str(item["column"])), + desc=item.get("direction", "asc") == "desc", + nulls_first=False, + ) + for item in node.config["order_by"] + ] + ), + ) + state.select_expressions = [ + exp.Star(), + window.as_(str(node.config["target_column"])), + ] + state.selected = True + + def _render_limit( node: GraphNode, state: _SqlRenderState, @@ -1642,7 +1738,10 @@ def _require_projection_slot( _node_sql_error( node.id, "sql.multiple_select", - "Only one expression, conversion, derive, select, or aggregate transform is supported in SQL view.", + ( + "Only one calculation, ranking, conversion, derive, " + "select, or aggregate transform is supported in SQL view." + ), ) ] ) @@ -1660,10 +1759,12 @@ def _register_sql_renderers() -> None: "select": _render_select, "derive": _render_derive, "expression": _render_expression, + "calculate": _render_calculate, "convert": _render_convert, "replace": _render_unsupported, "aggregate": _render_aggregate, "sort": _render_sort, + "window.rank": _render_rank, "limit": _render_limit, "quality.rules": _render_unsupported, "reconcile.compare": _render_unsupported, diff --git a/tests/test_node_library.py b/tests/test_node_library.py index db5f655..724fcfd 100644 --- a/tests/test_node_library.py +++ b/tests/test_node_library.py @@ -21,10 +21,12 @@ class DataflowNodeLibraryTests(unittest.TestCase): "select", "derive", "expression", + "calculate", "convert", "replace", "aggregate", "sort", + "window.rank", "limit", "quality.rules", "reconcile.compare", diff --git a/tests/test_operators.py b/tests/test_operators.py index 07863cd..b15b3af 100644 --- a/tests/test_operators.py +++ b/tests/test_operators.py @@ -57,6 +57,196 @@ class DataflowOperatorTests(unittest.TestCase): with self.assertRaises(ExpressionError): parse_expression("(select secret from credentials)") + def test_expression_vocabulary_covers_conditional_text_and_dates(self) -> None: + row = { + "code": "A-007", + "name": " Ada 42 ", + "amount": 12, + "created_on": "2026-07-30", + "started_on": "2026-07-01", + } + + self.assertEqual( + "00007", + evaluate_expression( + "lpad(split_part(code, '-', 2), 5, '0')", + row, + ), + ) + self.assertEqual( + "Ada", + evaluate_expression( + "regexp_replace(name, '[^A-Za-z]', '', 'g')", + row, + ), + ) + self.assertTrue( + evaluate_expression( + "code like 'A-%' and amount between 10 and 20", + row, + ) + ) + self.assertEqual( + "30.07.2026", + evaluate_expression( + "to_char(created_on, 'DD.MM.YYYY')", + row, + ), + ) + self.assertEqual( + 29, + evaluate_expression( + "date_diff('day', started_on, created_on)", + row, + ), + ) + + def test_ordered_calculations_and_partitioned_rank_execute(self) -> None: + graph = PipelineGraph( + nodes=[ + node( + "source", + "source.inline", + { + "source_name": "records", + "rows": [ + {"group": "A", "amount": 10}, + {"group": "A", "amount": 20}, + {"group": "A", "amount": 20}, + {"group": "B", "amount": 5}, + ], + }, + x=0, + ), + node( + "calculate", + "calculate", + { + "calculations": [ + { + "target_column": "gross", + "expression": "amount * 2", + "result_type": "integer", + }, + { + "target_column": "band", + "expression": ( + "case when gross >= 40 then 'high' " + "else 'standard' end" + ), + "result_type": "string", + }, + ] + }, + x=200, + ), + node( + "rank", + "window.rank", + { + "method": "rank", + "target_column": "group_rank", + "partition_by": ["group"], + "order_by": [ + {"column": "gross", "direction": "desc"} + ], + }, + x=400, + ), + node("output", "output", {}, x=600), + ], + edges=[ + GraphEdge(id="e1", source="source", target="calculate"), + GraphEdge(id="e2", source="calculate", target="rank"), + GraphEdge(id="e3", source="rank", target="output"), + ], + ) + + self.assertFalse( + [item for item in validate_graph(graph) if item.severity == "error"] + ) + result = execute_preview(graph, row_limit=100) + + self.assertEqual([3, 1, 1, 1], [ + item["group_rank"] for item in result.rows + ]) + self.assertEqual( + ["standard", "high", "high", "standard"], + [item["band"] for item in result.rows], + ) + + def test_semi_and_anti_joins_return_left_rows_once(self) -> None: + for join_type, expected in ( + ("semi", [{"id": "A"}, {"id": "C"}]), + ("anti", [{"id": "B"}]), + ): + with self.subTest(join_type=join_type): + graph = PipelineGraph( + nodes=[ + node( + "left", + "source.inline", + { + "source_name": "left_records", + "rows": [{"id": "A"}, {"id": "B"}, {"id": "C"}], + }, + x=0, + ), + node( + "right", + "source.inline", + { + "source_name": "right_records", + "rows": [ + {"id": "A"}, + {"id": "A"}, + {"id": "C"}, + ], + }, + x=0, + ), + node( + "join", + "combine.join", + { + "join_type": join_type, + "left_keys": ["id"], + "right_keys": ["id"], + "right_prefix": "right_", + }, + x=200, + ), + node("output", "output", {}, x=400), + ], + edges=[ + GraphEdge( + id="e1", + source="left", + target="join", + target_port="left", + ), + GraphEdge( + id="e2", + source="right", + target="join", + target_port="right", + ), + GraphEdge(id="e3", source="join", target="output"), + ], + ) + + self.assertFalse( + [ + item + for item in validate_graph(graph) + if item.severity == "error" + ] + ) + self.assertEqual( + expected, + execute_preview(graph, row_limit=100).rows, + ) + def test_conversion_expression_and_quality_nodes_execute(self) -> None: graph = PipelineGraph( nodes=[ @@ -362,6 +552,89 @@ class DataflowOperatorTests(unittest.TestCase): self.assertIn("LOWER(name) AS normalized_name", sql) + def test_calculation_and_rank_nodes_render_as_sql(self) -> None: + for transform, expected_sql in ( + ( + node( + "calculate", + "calculate", + { + "calculations": [ + { + "target_column": "gross", + "expression": "amount * 2", + "result_type": "integer", + }, + { + "target_column": "band", + "expression": ( + "case when amount >= 10 then 'high' " + "else 'standard' end" + ), + "result_type": "string", + }, + ] + }, + x=200, + ), + ("amount * 2 AS gross", "CASE WHEN amount >= 10"), + ), + ( + node( + "rank", + "window.rank", + { + "method": "row_number", + "target_column": "position", + "partition_by": ["group"], + "order_by": [ + {"column": "amount", "direction": "desc"} + ], + }, + x=200, + ), + ( + "ROW_NUMBER() OVER", + "PARTITION BY group ORDER BY amount DESC", + ), + ), + ): + with self.subTest(node_type=transform.type): + graph = PipelineGraph( + nodes=[ + node( + "source", + "source.inline", + { + "source_name": "records", + "rows": [ + {"group": "A", "amount": 12} + ], + }, + x=0, + ), + transform, + node("output", "output", {}, x=400), + ], + edges=[ + GraphEdge( + id="e1", + source="source", + target=transform.id, + ), + GraphEdge( + id="e2", + source=transform.id, + target="output", + ), + ], + ) + + sql, _ = render_sql(graph) + + for fragment in expected_sql: + self.assertIn(fragment, sql) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_typed_execution.py b/tests/test_typed_execution.py index 4cb646e..32b7c84 100644 --- a/tests/test_typed_execution.py +++ b/tests/test_typed_execution.py @@ -164,7 +164,46 @@ class ExecutionBackendTests(unittest.TestCase): DuckDbExecutionBackend().available(), "analytics extra is not installed", ) - def test_duckdb_matches_reference_for_aggregate_join_and_union(self) -> None: + def test_duckdb_matches_reference_for_supported_operators(self) -> None: + calculation_graph = simple_graph() + calculation_graph.nodes[1] = calculation_graph.nodes[1].model_copy( + update={ + "type": "calculate", + "label": "Calculated fields", + "config": { + "calculations": [ + { + "target_column": "double_amount", + "expression": "amount * 2", + "result_type": "integer", + }, + { + "target_column": "amount_band", + "expression": ( + "case when amount >= 10 then 'high' " + "else 'standard' end" + ), + "result_type": "string", + }, + ] + }, + } + ) + rank_graph = simple_graph() + rank_graph.nodes[1] = rank_graph.nodes[1].model_copy( + update={ + "type": "window.rank", + "label": "Department rank", + "config": { + "method": "rank", + "target_column": "department_rank", + "partition_by": ["department"], + "order_by": [ + {"column": "amount", "direction": "desc"} + ], + }, + } + ) graphs = [ compile_sql( """ @@ -214,6 +253,42 @@ class ExecutionBackendTests(unittest.TestCase): ), ], )[0], + compile_sql( + """ + SELECT records.department, records.amount + FROM records + SEMI JOIN labels + ON records.department = labels.department + ORDER BY records.amount + """, + source_nodes=[ + inline_source(), + inline_source( + "labels-source", + "labels", + [{"department": "a"}], + ), + ], + )[0], + compile_sql( + """ + SELECT records.department, records.amount + FROM records + ANTI JOIN labels + ON records.department = labels.department + ORDER BY records.amount + """, + source_nodes=[ + inline_source(), + inline_source( + "labels-source", + "labels", + [{"department": "a"}], + ), + ], + )[0], + calculation_graph, + rank_graph, ] for graph in graphs: with self.subTest(nodes=[node.type for node in graph.nodes]): diff --git a/webui/src/features/dataflow/NodeInspector.tsx b/webui/src/features/dataflow/NodeInspector.tsx index 3465869..e7b465d 100644 --- a/webui/src/features/dataflow/NodeInspector.tsx +++ b/webui/src/features/dataflow/NodeInspector.tsx @@ -38,7 +38,9 @@ export default function NodeInspector({ }: NodeInspectorProps) { const [rowsText, setRowsText] = useState(""); const [aggregateText, setAggregateText] = useState(""); + const [calculationText, setCalculationText] = useState(""); const [sortText, setSortText] = useState(""); + const [rankSortText, setRankSortText] = useState(""); const [rulesText, setRulesText] = useState(""); const [parametersText, setParametersText] = useState(""); const [subflowGraphText, setSubflowGraphText] = useState(""); @@ -47,7 +49,9 @@ export default function NodeInspector({ useEffect(() => { setRowsText(node ? JSON.stringify(node.config.rows ?? [], null, 2) : ""); setAggregateText(node ? aggregatesToText(node.config.aggregates) : ""); + setCalculationText(node ? calculationsToText(node.config.calculations) : ""); setSortText(node ? sortFieldsToText(node.config.fields) : ""); + setRankSortText(node ? sortFieldsToText(node.config.order_by) : ""); setRulesText(node ? JSON.stringify(node.config.rules ?? [], null, 2) : ""); setParametersText(node ? JSON.stringify(node.config.parameters ?? {}, null, 2) : ""); setSubflowGraphText(node ? JSON.stringify(node.config.graph ?? {}, null, 2) : ""); @@ -103,6 +107,26 @@ export default function NodeInspector({ } }; + const commitCalculations = () => { + try { + const parsed = calculationsFromText(calculationText); + setLocalError(""); + updateConfig({ calculations: parsed }); + } catch (error) { + setLocalError(error instanceof Error ? error.message : "Calculations could not be parsed."); + } + }; + + const commitRankSort = () => { + try { + const parsed = sortFieldsFromText(rankSortText); + setLocalError(""); + updateConfig({ order_by: parsed }); + } catch (error) { + setLocalError(error instanceof Error ? error.message : "Rank order could not be parsed."); + } + }; + const commitJsonConfig = ( field: string, value: string, @@ -318,6 +342,8 @@ export default function NodeInspector({ + + @@ -334,13 +360,15 @@ export default function NodeInspector({ disabled={readOnly} /> - - updateConfig({ right_prefix: event.target.value })} - disabled={readOnly} - /> - + {!["semi", "anti"].includes(textValue(node.config.join_type)) ? ( + + updateConfig({ right_prefix: event.target.value })} + disabled={readOnly} + /> + + ) : null} ) : null} {node.type === "select" ? ( @@ -453,6 +481,18 @@ export default function NodeInspector({ ) : null} + {node.type === "calculate" ? ( + +