feat(expr-ir): Support over(*partition_by)

Child of #2572

Author: dangotbanned

narwhals/_plan/arrow/dataframe.py

@@ -152,7 +152,7 @@ def join(
         *,
         how: NonCrossJoinStrategy,
         left_on: Sequence[str],
-        right_on: Sequence[str],
+        right_on: Sequence[str] = (),
         suffix: str = "_right",
     ) -> Self:
         left, right = self.native, other.native

narwhals/_plan/arrow/expr.py

@@ -322,13 +322,27 @@ def min(self, node: Min, frame: Frame, name: str) -> Scalar:
         return self._with_native(result, name)
 
     # TODO @dangotbanned: top-level, complex-ish nodes
-    # - [ ] `over`/`_ordered` (with partitions) requires `group_by`, `join`
-    # - [x] `over_ordered` alone should be possible w/ the current API
-    # - [x] `map_batches` is defined in `EagerExpr`, might be simpler here than on main
+    # - [ ] Over
+    #   - [x] `over_ordered`
+    #   - [x] `group_by`, `join`
+    #   - [!] `over`
+    #   - [ ] `over_ordered` (with partitions)
+    # - [ ] `map_batches`
+    #   - [x] elementwise
+    #   - [ ] scalar
     # - [ ] `rolling_expr` has 4 variants
 
     def over(self, node: ir.WindowExpr, frame: Frame, name: str) -> Self:
-        raise NotImplementedError
+        expr = node.expr
+        resolved = frame._grouper.by_irs(*node.partition_by).agg_irs(expr).resolve(frame)
+        by_names = resolved.key_names
+        result = (
+            frame.select_names(*by_names)
+            .join(resolved.evaluate(frame), how="left", left_on=by_names)
+            .get_column(expr.meta.output_name())
+            .native
+        )
+        return self._with_native(result, name)
 
     def over_ordered(
         self, node: ir.OrderedWindowExpr, frame: Frame, name: str

narwhals/_plan/compliant/group_by.py

@@ -203,3 +203,13 @@ class Grouped(Grouper[Resolved]):
     @property
     def _resolver(self) -> type[Resolved]:
         return Resolved
+
+    @classmethod
+    def by_irs(cls, *by: ExprIR) -> Self:
+        obj = cls.__new__(cls)
+        obj._keys = by
+        return obj
+
+    def agg_irs(self, *aggs: ExprIR) -> Self:
+        self._aggs = aggs
+        return self