feat(dataflow): model production SQL flow patterns

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