feat(dataflow): model production SQL flow patterns

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
+arithmetic, comparisons, boolean logic, `CASE`, safe casts, predicates,
-string/numeric function catalogue without Python evaluation or effectful SQL.
+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
+The executable fixtures in `fixtures/golden` cover monthly structured-file
-reconciliation story and the sanctions-screening story with reviewable sample
+reconciliation, sanctions screening, a HEICO-style current-status export, and
-inputs and exact expected outputs.
+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
 

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).

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.

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"
+  }
+]

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"}
+  ]
+}

fixtures/golden/heico-student-status/inputs/export-exclusions.json

@@ -0,0 +1,6 @@
+[
+  {
+    "student_id": "S-200",
+    "reason": "No active enrollment"
+  }
+]

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"
+  }
+]

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"
+}

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
+  }
+]

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"}
+  ]
+}

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
+  }
+]

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"
+}

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"])]
         ),

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,

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",

src/govoplan_dataflow/backend/ir.py

@@ -279,16 +279,34 @@ 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 (
+    return IrExpression(
-        IrExpression(
         source=parsed.source,
         columns=parsed.columns,
         result_type=_data_type(inferred),
@@ -299,7 +317,6 @@ def _node_expressions(
                 "result_type": inferred,
             }
         ),
-        ),
     )
 
 

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",

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,

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."
+                    ),
+                )
+            ]
+        )
+    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 "")
     )
-    join_type = 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,

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",

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()

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]):

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({
                 <option value="left">All left rows</option>
                 <option value="right">All right rows</option>
                 <option value="full">All rows</option>
+                <option value="semi">Left rows with a match</option>
+                <option value="anti">Left rows without a match</option>
               </select>
             </FormField>
             <FormField label="Left keys">
@@ -334,6 +360,7 @@ export default function NodeInspector({
                 disabled={readOnly}
               />
             </FormField>
+            {!["semi", "anti"].includes(textValue(node.config.join_type)) ? (
               <FormField label="Right-column prefix">
                 <input
                   value={textValue(node.config.right_prefix)}
@@ -341,6 +368,7 @@ export default function NodeInspector({
                   disabled={readOnly}
                 />
               </FormField>
+            ) : null}
           </>
         ) : null}
         {node.type === "select" ? (
@@ -453,6 +481,18 @@ export default function NodeInspector({
             </FormField>
           </>
         ) : null}
+        {node.type === "calculate" ? (
+          <FormField label="Calculated columns">
+            <textarea
+              className="dataflow-expression-editor"
+              value={calculationText}
+              onChange={(event) => setCalculationText(event.target.value)}
+              onBlur={commitCalculations}
+              spellCheck={false}
+              disabled={readOnly}
+            />
+          </FormField>
+        ) : null}
         {node.type === "convert" ? (
           <>
             <FormField label="Source column">
@@ -550,6 +590,45 @@ export default function NodeInspector({
             />
           </FormField>
         ) : null}
+        {node.type === "window.rank" ? (
+          <>
+            <FormField label="Method">
+              <select
+                value={textValue(node.config.method) || "row_number"}
+                onChange={(event) => updateConfig({ method: event.target.value })}
+                disabled={readOnly}
+              >
+                <option value="row_number">Row number</option>
+                <option value="rank">Rank with gaps</option>
+                <option value="dense_rank">Dense rank</option>
+              </select>
+            </FormField>
+            <FormField label="Output column">
+              <input
+                value={textValue(node.config.target_column)}
+                onChange={(event) => updateConfig({ target_column: event.target.value })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Partition by">
+              <input
+                value={stringList(node.config.partition_by).join(", ")}
+                onChange={(event) => updateConfig({ partition_by: commaList(event.target.value) })}
+                disabled={readOnly}
+              />
+            </FormField>
+            <FormField label="Order by">
+              <textarea
+                className="dataflow-expression-editor"
+                value={rankSortText}
+                onChange={(event) => setRankSortText(event.target.value)}
+                onBlur={commitRankSort}
+                spellCheck={false}
+                disabled={readOnly}
+              />
+            </FormField>
+          </>
+        ) : null}
         {node.type === "limit" ? (
           <FormField label="Maximum rows">
             <input
@@ -719,6 +798,51 @@ function selectFieldsFromText(value: string): Array<{ column: string; alias: str
   });
 }
 
+function calculationsToText(value: unknown): string {
+  if (!Array.isArray(value)) return "";
+  return value.map((item) => {
+    if (!isRecord(item)) return "";
+    const target = textValue(item.target_column);
+    const expression = textValue(item.expression);
+    const resultType = textValue(item.result_type) || "unknown";
+    const typeSuffix = resultType === "unknown" ? "" : `:${resultType}`;
+    return target && expression ? `${target}${typeSuffix} = ${expression}` : "";
+  }).filter(Boolean).join("\n");
+}
+
+function calculationsFromText(
+  value: string
+): Array<{ target_column: string; expression: string; result_type: string }> {
+  const lines = value.split("\n").map((line) => line.trim()).filter(Boolean);
+  if (!lines.length) throw new Error("Add at least one calculated column.");
+  const supportedTypes = new Set([
+    "unknown",
+    "string",
+    "integer",
+    "number",
+    "boolean",
+    "date",
+    "datetime"
+  ]);
+  return lines.map((line) => {
+    const separator = line.indexOf("=");
+    if (separator < 1) throw new Error(`Invalid calculation: ${line}`);
+    const declaration = line.slice(0, separator).trim();
+    const expression = line.slice(separator + 1).trim();
+    const [targetColumn, declaredType = "unknown"] = declaration
+      .split(":", 2)
+      .map((item) => item.trim());
+    if (!targetColumn || !expression || !supportedTypes.has(declaredType)) {
+      throw new Error(`Invalid calculation: ${line}`);
+    }
+    return {
+      target_column: targetColumn,
+      expression,
+      result_type: declaredType
+    };
+  });
+}
+
 function aggregatesToText(value: unknown): string {
   if (!Array.isArray(value)) return "";
   return value.map((item) => {
@@ -745,7 +869,8 @@ function sortFieldsToText(value: unknown): string {
   if (!Array.isArray(value)) return "";
   return value.map((item) => {
     if (!isRecord(item)) return "";
-    return `${textValue(item.column)} ${textValue(item.direction) || "asc"}`;
+    const column = textValue(item.column);
+    return column ? `${column} ${textValue(item.direction) || "asc"}` : "";
   }).filter(Boolean).join("\n");
 }
 

webui/src/features/dataflow/model.ts

@@ -140,6 +140,21 @@ export const FALLBACK_NODE_LIBRARY: NodeTypeDefinition[] = [
     output,
     { target_column: "", expression: "", result_type: "unknown" }
   ),
+  nodeType(
+    "calculate",
+    "transform",
+    "Transform",
+    "Calculate columns",
+    "Create or replace several ordered calculated fields.",
+    "calculator",
+    input,
+    output,
+    {
+      calculations: [
+        { target_column: "", expression: "", result_type: "unknown" }
+      ]
+    }
+  ),
   nodeType(
     "convert",
     "transform",
@@ -184,6 +199,22 @@ export const FALLBACK_NODE_LIBRARY: NodeTypeDefinition[] = [
     output,
     { fields: [{ column: "", direction: "asc" }] }
   ),
+  nodeType(
+    "window.rank",
+    "transform",
+    "Transform",
+    "Rank rows",
+    "Number or rank rows within partitions.",
+    "list-ordered",
+    input,
+    output,
+    {
+      method: "row_number",
+      target_column: "row_number",
+      partition_by: [],
+      order_by: [{ column: "", direction: "asc" }]
+    }
+  ),
   nodeType("limit", "transform", "Transform", "Limit rows", "Keep the first rows.", "list-end", input, output, {
     count: 100
   }),

webui/src/features/dataflow/nodeIcons.ts

@@ -3,6 +3,7 @@ import {
   BadgeCheck,
   Boxes,
   Braces,
+  Calculator,
   Columns3,
   Combine,
   Database,
@@ -11,6 +12,7 @@ import {
   ListEnd,
   ListFilter,
   ListChecks,
+  ListOrdered,
   PanelTopOpen,
   Replace,
   ReplaceAll,
@@ -26,6 +28,7 @@ const icons: Record<string, LucideIcon> = {
   "badge-check": BadgeCheck,
   boxes: Boxes,
   braces: Braces,
+  calculator: Calculator,
   "columns-3": Columns3,
   combine: Combine,
   database: Database,
@@ -34,6 +37,7 @@ const icons: Record<string, LucideIcon> = {
   "list-end": ListEnd,
   "list-filter": ListFilter,
   "list-checks": ListChecks,
+  "list-ordered": ListOrdered,
   "panel-top-open": PanelTopOpen,
   replace: Replace,
   "replace-all": ReplaceAll,

webui/tsconfig.json

@@ -26,7 +26,8 @@
       "@xyflow/react": ["../../govoplan-core/webui/node_modules/@xyflow/react/dist/esm/index.d.ts"],
       "lucide-react": ["../../govoplan-core/webui/node_modules/lucide-react/dist/lucide-react.d.ts"],
       "react": ["../../govoplan-core/webui/node_modules/@types/react/index.d.ts"],
-      "react/jsx-runtime": ["../../govoplan-core/webui/node_modules/@types/react/jsx-runtime.d.ts"]
+      "react/jsx-runtime": ["../../govoplan-core/webui/node_modules/@types/react/jsx-runtime.d.ts"],
+      "react-router-dom": ["../../govoplan-core/webui/node_modules/react-router-dom/dist/index.d.ts"]
     }
   },
   "include": ["src"]