refactor(expr-ir): Make arrow.functions a package

narwhals/_plan/arrow/acero.py

@@ -25,6 +25,7 @@
 import pyarrow.compute as pc  # ignore-banned-import
 from pyarrow.acero import Declaration as Decl
 
+from narwhals._plan.arrow.guards import is_expression
 from narwhals._plan.common import ensure_list_str, temp
 from narwhals._plan.typing import NonCrossJoinStrategy, OneOrSeq
 from narwhals._utils import check_column_names_are_unique
@@ -40,7 +41,7 @@
         Sequence,
     )
 
-    from typing_extensions import TypeAlias, TypeIs
+    from typing_extensions import TypeAlias
 
     from narwhals._arrow.typing import (  # type: ignore[attr-defined]
         AggregateOptions as _AggregateOptions,
@@ -117,12 +118,8 @@ def cols_iter(names: Iterable[str], /) -> Iterator[Expr]:
         yield col(name)
 
 
-def _is_expr(obj: Any) -> TypeIs[pc.Expression]:
-    return isinstance(obj, pc.Expression)
-
-
 def _parse_into_expr(into: IntoExpr, /, *, str_as_lit: bool = False) -> Expr:
-    if _is_expr(into):
+    if is_expression(into):
         return into
     if isinstance(into, str) and not str_as_lit:
         return col(into)
@@ -223,7 +220,7 @@ def project(**named_exprs: IntoExpr) -> Decl:
 
 
 def _add_column(native: pa.Table, index: int, name: str, values: IntoExpr) -> Decl:
-    column = values if _is_expr(values) else lit(values)
+    column = values if is_expression(values) else lit(values)
     schema = native.schema
     schema_names = schema.names
     if index == 0:
@@ -323,8 +320,8 @@ def _join_asof_strategy_to_tolerance(
     if strategy == "nearest":
         msg = "Only 'backward' and 'forward' strategies are currently supported for `pyarrow`"
         raise NotImplementedError(msg)
-    lower = fn.min_horizontal(fn.min_(left_on), fn.min_(right_on))
-    upper = fn.max_horizontal(fn.max_(left_on), fn.max_(right_on))
+    lower = fn.min_horizontal(fn.min(left_on), fn.min(right_on))
+    upper = fn.max_horizontal(fn.max(left_on), fn.max(right_on))
     scalar = fn.sub(lower, upper) if strategy == "backward" else fn.sub(upper, lower)
     tolerance: int = fn.cast(scalar, fn.I64).as_py()
     return tolerance

narwhals/_plan/arrow/common.py

@@ -2,27 +2,22 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, ClassVar, Generic
+from typing import TYPE_CHECKING, ClassVar, Generic
 
 from narwhals._plan.arrow import compat
 from narwhals._plan.arrow.functions import random_indices
+from narwhals._plan.arrow.guards import is_series
 from narwhals._typing_compat import TypeVar
 from narwhals._utils import Implementation, Version, _StoresNative
 
 if TYPE_CHECKING:
     import pyarrow as pa
-    from typing_extensions import Self, TypeIs
+    from typing_extensions import Self
 
     from narwhals._plan.arrow.namespace import ArrowNamespace
     from narwhals._plan.arrow.typing import ChunkedArrayAny, Indices
 
 
-def is_series(obj: Any) -> TypeIs[_StoresNative[ChunkedArrayAny]]:
-    from narwhals._plan.arrow.series import ArrowSeries
-
-    return isinstance(obj, ArrowSeries)
-
-
 NativeT = TypeVar("NativeT", "pa.Table", "ChunkedArrayAny")
 
 

narwhals/_plan/arrow/dataframe.py

@@ -12,7 +12,11 @@
 from narwhals._plan.arrow import acero, compat, functions as fn
 from narwhals._plan.arrow.common import ArrowFrameSeries as FrameSeries
 from narwhals._plan.arrow.expr import ArrowExpr as Expr, ArrowScalar as Scalar
-from narwhals._plan.arrow.group_by import ArrowGroupBy as GroupBy, partition_by
+from narwhals._plan.arrow.group_by import (
+    ArrowGroupBy as GroupBy,
+    partition_by,
+    unique_keep_boolean_length_preserving,
+)
 from narwhals._plan.arrow.pivot import pivot_table
 from narwhals._plan.arrow.series import ArrowSeries as Series
 from narwhals._plan.common import temp
@@ -142,7 +146,7 @@ def _unique(
         [`unsort_indices`]: https://github.com/narwhals-dev/narwhals/blob/9b9122b4ab38a6aebe2f09c29ad0f6191952a7a7/narwhals/_plan/arrow/functions.py#L1666-L1697
         """
         subset = tuple(subset or self.columns)
-        into_column_agg, mask = fn.unique_keep_boolean_length_preserving(keep)
+        into_column_agg, mask = unique_keep_boolean_length_preserving(keep)
         idx_name = temp.column_name(self.columns)
         df = self.select_names(*set(subset).union(order_by))
         if order_by:
@@ -227,7 +231,7 @@ def to_struct(self, name: str = "") -> Series:
             else:
                 struct = fn.chunked_array([], pa.struct(native.schema))
         else:
-            struct = fn.struct(native.column_names, native.columns)
+            struct = fn.struct.into_struct(native.columns, native.column_names)
         return Series.from_native(struct, name, version=self.version)
 
     def get_column(self, name: str) -> Series: