feat(expr-ir): Acero order_by, hashjoin , DataFrame.{filter,join}, Expr.is_{first,last}_distinct

narwhals/_plan/arrow/acero.py

@@ -25,6 +25,8 @@
 import pyarrow.compute as pc  # ignore-banned-import
 from pyarrow.acero import Declaration as Decl
 
+from narwhals._plan.common import flatten_hash_safe
+from narwhals._plan.options import SortMultipleOptions
 from narwhals._plan.typing import OneOrSeq
 from narwhals.typing import SingleColSelector
 
@@ -189,10 +191,15 @@ def _order_by(
     return Decl("order_by", pac.OrderByNodeOptions(keys, null_placement=null_placement))
 
 
-# TODO @dangotbanned: Utilize `SortMultipleOptions.to_arrow_acero`
-def sort_by(*args: Any, **kwds: Any) -> Decl:
-    msg = "Should convert from polars args -> use `_order_by"
-    raise NotImplementedError(msg)
+def sort_by(
+    by: OneOrIterable[str],
+    *more_by: str,
+    descending: OneOrIterable[bool] = False,
+    nulls_last: bool = False,
+) -> Decl:
+    return SortMultipleOptions.parse(
+        descending=descending, nulls_last=nulls_last
+    ).to_arrow_acero(tuple(flatten_hash_safe((by, more_by))))
 
 
 def collect(*declarations: Decl, use_threads: bool = True) -> pa.Table:

narwhals/_plan/arrow/expr.py

@@ -9,18 +9,13 @@
 from narwhals._plan.arrow import functions as fn
 from narwhals._plan.arrow.series import ArrowSeries as Series
 from narwhals._plan.arrow.typing import ChunkedOrScalarAny, NativeScalar, StoresNativeT_co
+from narwhals._plan.common import temp
 from narwhals._plan.compliant.column import ExprDispatch
 from narwhals._plan.compliant.expr import EagerExpr
 from narwhals._plan.compliant.scalar import EagerScalar
 from narwhals._plan.compliant.typing import namespace
 from narwhals._plan.expressions import NamedIR
-from narwhals._utils import (
-    Implementation,
-    Version,
-    _StoresNative,
-    generate_temporary_column_name,
-    not_implemented,
-)
+from narwhals._utils import Implementation, Version, _StoresNative, not_implemented
 from narwhals.exceptions import InvalidOperationError, ShapeError
 
 if TYPE_CHECKING:
@@ -231,10 +226,8 @@ def sort(self, node: ir.Sort, frame: Frame, name: str) -> Expr:
 
     def sort_by(self, node: ir.SortBy, frame: Frame, name: str) -> Expr:
         series = self._dispatch_expr(node.expr, frame, name)
-        by = (
-            self._dispatch_expr(e, frame, f"<TEMP>_{idx}")
-            for idx, e in enumerate(node.by)
-        )
+        it_names = temp.column_names(frame)
+        by = (self._dispatch_expr(e, frame, nm) for e, nm in zip(node.by, it_names))
         df = namespace(self)._concat_horizontal((series, *by))
         names = df.columns[1:]
         indices = pc.sort_indices(df.native, options=node.options.to_arrow(names))
@@ -342,7 +335,7 @@ def over_ordered(
         # NOTE: Converting `over(order_by=..., options=...)` into the right shape for `DataFrame.sort`
         sort_by = tuple(NamedIR.from_ir(e) for e in node.order_by)
         options = node.sort_options.to_multiple(len(node.order_by))
-        idx_name = generate_temporary_column_name(8, frame.columns)
+        idx_name = temp.column_name(frame)
         sorted_context = frame.with_row_index(idx_name).sort(sort_by, options)
         evaluated = node.expr.dispatch(self, sorted_context.drop([idx_name]), name)
         if isinstance(evaluated, ArrowScalar):

narwhals/_plan/common.py

@@ -22,6 +22,8 @@
 
     from typing_extensions import TypeIs
 
+    from narwhals._plan.compliant.series import CompliantSeries
+    from narwhals._plan.series import Series
     from narwhals._plan.typing import (
         DTypeT,
         ExprIRT,
@@ -109,9 +111,21 @@ def into_dtype(dtype: DTypeT | type[NonNestedDTypeT], /) -> DTypeT | NonNestedDT
     return dtype
 
 
-# TODO @dangotbanned: Review again and try to work around (https://github.com/microsoft/pyright/issues/10673#issuecomment-3033789021)
+# NOTE: See (https://github.com/microsoft/pyright/issues/10673#issuecomment-3033789021)
 # The issue is `T` possibly being `Iterable`
 # Ignoring here still leaks the issue to the caller, where you need to annotate the base case
+@overload
+def flatten_hash_safe(iterable: Iterable[OneOrIterable[str]], /) -> Iterator[str]: ...
+@overload
+def flatten_hash_safe(
+    iterable: Iterable[OneOrIterable[Series]], /
+) -> Iterator[Series]: ...
+@overload
+def flatten_hash_safe(
+    iterable: Iterable[OneOrIterable[CompliantSeries]], /
+) -> Iterator[CompliantSeries]: ...
+@overload
+def flatten_hash_safe(iterable: Iterable[OneOrIterable[T]], /) -> Iterator[T]: ...
 def flatten_hash_safe(iterable: Iterable[OneOrIterable[T]], /) -> Iterator[T]:
     """Fully unwrap all levels of nesting.
 

narwhals/_plan/expressions/expr.py

@@ -143,8 +143,7 @@ class Exclude(_ColumnSelection, child=("expr",)):
 
     @staticmethod
     def from_names(expr: ExprIR, *names: str | t.Iterable[str]) -> Exclude:
-        flat: t.Iterator[str] = flatten_hash_safe(names)
-        return Exclude(expr=expr, names=tuple(flat))
+        return Exclude(expr=expr, names=tuple(flatten_hash_safe(names)))
 
     def __repr__(self) -> str:
         return f"{self.expr!r}.exclude({list(self.names)!r})"

narwhals/_plan/expressions/selectors.py

@@ -14,7 +14,6 @@
 from narwhals._utils import Version, _parse_time_unit_and_time_zone
 
 if TYPE_CHECKING:
-    from collections.abc import Iterator
     from datetime import timezone
     from typing import TypeVar
 
@@ -127,7 +126,7 @@ def from_string(pattern: str, /) -> Matches:
     @staticmethod
     def from_names(*names: OneOrIterable[str]) -> Matches:
         """Implements `cs.by_name` to support `__r<op>__` with column selections."""
-        it: Iterator[str] = flatten_hash_safe(names)
+        it = flatten_hash_safe(names)
         return Matches.from_string(f"^({'|'.join(re.escape(name) for name in it)})$")
 
     def __repr__(self) -> str: