refactor: Dispatch IO functions via *Namespace methods

docs/extending.md

@@ -59,7 +59,54 @@ handle plugins. For this integration to work, any plugin architecture must conta
 
     Take a look at the `Plugin` protocol in `narwhals/plugins.py` for the
     signatures.
-
+
+## IO functions: the namespace contract
+
+The Narwhals IO functions (`read_csv`, `scan_csv`, `read_parquet`, `scan_parquet`)
+dispatch to same-named methods on the compliant namespace. This is a single mechanism,
+shared by built-in backends and extensions alike: to support these functions, a
+compliant namespace implements (a subset of):
+
+```py
+from narwhals.typing import NormalizedPath
+
+
+def read_csv(
+    self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+) -> CompliantDataFrame:
+    ...
+
+
+def scan_csv(
+    self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+) -> CompliantFrame:
+    ...
+
+
+def read_parquet(self, source: NormalizedPath, **kwds: Any) -> CompliantDataFrame:
+    ...
+
+
+def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> CompliantFrame:
+    ...
+```
+
+In all cases:
+
+- `source` is a plain string at runtime: `NormalizedPath` is a `str`
+  [`NewType`](https://docs.python.org/3/library/typing.html#newtype) tagging that
+  Narwhals has already normalized `Path` and path-like inputs before dispatching to the
+  namespace.
+- `kwds` are forwarded to the native reader, and it is the namespace's responsibility
+  to translate `separator` into whatever its native CSV reader expects (and to raise if
+  the two conflict).
+- `read_*` methods are eager-only, so they are only ever called on namespaces of eager
+  backends. `scan_*` methods are called for any backend: lazy namespaces return a
+  compliant LazyFrame, while eager ones may simply read eagerly. Namespaces complying
+  with the `EagerNamespace` protocol only need to implement `read_csv` and
+  `read_parquet`, as they inherit `scan_*` default implementations which fall back to
+  the corresponding `read_*` method.
+
 ## Can I see an example?
 
 Yes! For a reference plugin, please check out [narwhals-daft](https://github.com/narwhals-dev/narwhals-daft).

packages/test-plugin/src/test_plugin/namespace.py

@@ -37,3 +37,5 @@ def from_native(self, native_object: DictFrame) -> DictLazyFrame:
     selectors: Any = not_implemented()
     coalesce: Any = not_implemented()
     struct: Any = not_implemented()
+    scan_csv: Any = not_implemented()
+    scan_parquet: Any = not_implemented()

src/narwhals/_arrow/namespace.py

@@ -3,7 +3,7 @@
 import operator
 from functools import reduce
 from itertools import chain
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 import pyarrow as pa
 import pyarrow.compute as pc
@@ -25,7 +25,12 @@
 
     from narwhals._arrow.typing import ChunkedArrayAny, Incomplete, ScalarAny
     from narwhals._utils import Version
-    from narwhals.typing import CorrelationMethod, IntoDType, PythonLiteral
+    from narwhals.typing import (
+        CorrelationMethod,
+        IntoDType,
+        NormalizedPath,
+        PythonLiteral,
+    )
 
 
 class ArrowNamespace(
@@ -48,6 +53,27 @@ def _series(self) -> type[ArrowSeries]:
     def __init__(self, *, version: Version) -> None:
         self._version = version
 
+    def read_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> ArrowDataFrame:
+        from pyarrow import csv
+
+        if (parse_options := kwds.get("parse_options")) is not None:
+            if cast("csv.ParseOptions", parse_options).delimiter != separator:
+                msg = (
+                    "`separator` and `parse_options.delimiter` do not match: "
+                    f"`separator`={separator} and `delimiter`={parse_options.delimiter}."
+                )
+                raise TypeError(msg)
+        else:
+            kwds["parse_options"] = csv.ParseOptions(delimiter=separator)
+        return self._dataframe.from_native(csv.read_csv(source, **kwds), context=self)
+
+    def read_parquet(self, source: NormalizedPath, **kwds: Any) -> ArrowDataFrame:
+        from pyarrow import parquet as pq
+
+        return self._dataframe.from_native(pq.read_table(source, **kwds), context=self)
+
     def extract_native(
         self, *series: ArrowSeries
     ) -> Iterator[ChunkedArrayAny | ScalarAny]:

src/narwhals/_compliant/namespace.py

@@ -45,6 +45,7 @@
         Into1DArray,
         IntoDType,
         NonNestedLiteral,
+        NormalizedPath,
         _2DArray,
     )
 
@@ -109,6 +110,10 @@ def list(self, *exprs: CompliantExprT) -> CompliantExprT: ...
     def selectors(self) -> CompliantSelectorNamespace[Any, Any]: ...
     def struct(self, *exprs: CompliantExprT) -> CompliantExprT: ...
     def coalesce(self, *exprs: CompliantExprT) -> CompliantExprT: ...
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> CompliantFrameT: ...
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> CompliantFrameT: ...
     # NOTE: typing this accurately requires 2x more `TypeVar`s
     def from_native(self, data: Any, /) -> Any: ...
     def is_native(self, obj: Any, /) -> TypeIs[Any]:
@@ -219,6 +224,18 @@ def _backend_version(self) -> tuple[int, ...]:
     def _dataframe(self) -> type[EagerDataFrameT]: ...
     @property
     def _series(self) -> type[EagerSeriesT_co]: ...
+    def read_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> EagerDataFrameT: ...
+    def read_parquet(self, source: NormalizedPath, **kwds: Any) -> EagerDataFrameT: ...
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> EagerDataFrameT:
+        return self.read_csv(source, separator=separator, **kwds)
+
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> EagerDataFrameT:
+        return self.read_parquet(source, **kwds)
+
     def _if_then_else(
         self,
         when: NativeSeriesT,

src/narwhals/_dask/namespace.py

@@ -4,7 +4,7 @@
 from datetime import date, datetime
 from functools import reduce
 from itertools import chain
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import dask.dataframe as dd
 import pandas as pd
@@ -22,7 +22,12 @@
     combine_alias_output_names,
     combine_evaluate_output_names,
 )
-from narwhals._utils import Implementation, is_nested_literal, not_implemented
+from narwhals._utils import (
+    Implementation,
+    is_nested_literal,
+    not_implemented,
+    validate_separators,
+)
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Iterator
@@ -35,6 +40,7 @@
         CorrelationMethod,
         IntoDType,
         NonNestedLiteral,
+        NormalizedPath,
     )
 
 
@@ -59,6 +65,16 @@ def _lazyframe(self) -> type[DaskLazyFrame]:
     def __init__(self, *, version: Version) -> None:
         self._version = version
 
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> DaskLazyFrame:
+        validate_separators(separator, ("sep",), kwds)
+        native = dd.read_csv(source, sep=separator, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> DaskLazyFrame:
+        return self._lazyframe.from_native(dd.read_parquet(source, **kwds), context=self)
+
     def lit(self, value: NonNestedLiteral, dtype: IntoDType | None) -> DaskExpr:
         if is_nested_literal(value):
             msg = f"Nested structures are not supported for Dask backend, found {type(value).__name__}"

src/narwhals/_duckdb/namespace.py

@@ -29,7 +29,7 @@
     evaluate_output_names_and_aliases,
 )
 from narwhals._sql.namespace import SQLNamespace
-from narwhals._utils import Implementation, requires
+from narwhals._utils import Implementation, requires, validate_separators
 
 if TYPE_CHECKING:
     from collections.abc import Callable, Iterable, Mapping
@@ -38,7 +38,13 @@
 
     from narwhals._compliant.window import WindowInputs
     from narwhals._utils import Version
-    from narwhals.typing import ConcatMethod, CorrelationMethod, IntoDType, PythonLiteral
+    from narwhals.typing import (
+        ConcatMethod,
+        CorrelationMethod,
+        IntoDType,
+        NormalizedPath,
+        PythonLiteral,
+    )
 
 VARCHAR = duckdb_dtypes.VARCHAR
 
@@ -63,6 +69,17 @@ def _expr(self) -> type[DuckDBExpr]:
     def _lazyframe(self) -> type[DuckDBLazyFrame]:
         return DuckDBLazyFrame
 
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> DuckDBLazyFrame:
+        validate_separators(separator, ("delimiter", "delim", "sep"), kwds)
+        native = duckdb.read_csv(source, delimiter=separator, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> DuckDBLazyFrame:
+        native = duckdb.read_parquet(source, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
     def _function(self, name: str, *args: Expression) -> Expression:  # type: ignore[override]
         return function(name, *args)
 

src/narwhals/_ibis/namespace.py

@@ -19,13 +19,19 @@
 from narwhals._ibis.selectors import IbisSelectorNamespace
 from narwhals._ibis.utils import function, lit, narwhals_to_native_dtype
 from narwhals._sql.namespace import SQLNamespace
-from narwhals._utils import Implementation
+from narwhals._utils import Implementation, validate_separators
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Mapping, Sequence
 
     from narwhals._utils import Version
-    from narwhals.typing import ConcatMethod, CorrelationMethod, IntoDType, PythonLiteral
+    from narwhals.typing import (
+        ConcatMethod,
+        CorrelationMethod,
+        IntoDType,
+        NormalizedPath,
+        PythonLiteral,
+    )
 
 
 class IbisNamespace(
@@ -49,6 +55,17 @@ def _expr(self) -> type[IbisExpr]:
     def _lazyframe(self) -> type[IbisLazyFrame]:
         return IbisLazyFrame
 
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> IbisLazyFrame:
+        validate_separators(separator, ("sep",), kwds)
+        native = ibis.read_csv(source, sep=separator, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> IbisLazyFrame:
+        native = ibis.read_parquet(source, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
     def _function(self, name: str, *args: ir.Value | PythonLiteral) -> ir.Value:
         return function(name, *args)
 

src/narwhals/_pandas_like/namespace.py

@@ -19,13 +19,19 @@
 from narwhals._pandas_like.series import PandasLikeSeries
 from narwhals._pandas_like.typing import NativeDataFrameT, NativeSeriesT
 from narwhals._pandas_like.utils import is_dtype_pyarrow, is_non_nullable_boolean
+from narwhals._utils import validate_separators
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
     from typing import TypeAlias
 
     from narwhals._utils import Implementation, Version
-    from narwhals.typing import CorrelationMethod, IntoDType, PythonLiteral
+    from narwhals.typing import (
+        CorrelationMethod,
+        IntoDType,
+        NormalizedPath,
+        PythonLiteral,
+    )
 
 
 Incomplete: TypeAlias = Any
@@ -69,6 +75,18 @@ def __init__(self, implementation: Implementation, version: Version) -> None:
         self._implementation = implementation
         self._version = version
 
+    def read_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> PandasLikeDataFrame:
+        validate_separators(separator, ("sep",), kwds)
+        ns = self._implementation.to_native_namespace()
+        native = ns.read_csv(source, sep=separator, **kwds)
+        return self._dataframe.from_native(native, context=self)
+
+    def read_parquet(self, source: NormalizedPath, **kwds: Any) -> PandasLikeDataFrame:
+        ns = self._implementation.to_native_namespace()
+        return self._dataframe.from_native(ns.read_parquet(source, **kwds), context=self)
+
     def coalesce(self, *exprs: PandasLikeExpr) -> PandasLikeExpr:
         def func(df: PandasLikeDataFrame) -> list[PandasLikeSeries]:
             series = (s for _expr in exprs for s in _expr(df))

src/narwhals/_polars/namespace.py

@@ -26,7 +26,7 @@
     from narwhals._polars.dataframe import Method, PolarsDataFrame, PolarsLazyFrame
     from narwhals._polars.typing import FrameT
     from narwhals._utils import _LimitedContext
-    from narwhals.typing import Into1DArray, IntoDType, TimeUnit, _2DArray
+    from narwhals.typing import Into1DArray, IntoDType, NormalizedPath, TimeUnit, _2DArray
 
 
 class PolarsNamespace:
@@ -115,6 +115,24 @@ def from_numpy(
             return self._dataframe.from_numpy(data, schema=schema, context=self)
         return self._series.from_numpy(data, context=self)  # pragma: no cover
 
+    def read_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> PolarsDataFrame:
+        native = pl.read_csv(source, separator=separator, **kwds)
+        return self._dataframe.from_native(native, context=self)
+
+    def scan_csv(
+        self, source: NormalizedPath, *, separator: str = ",", **kwds: Any
+    ) -> PolarsLazyFrame:
+        native = pl.scan_csv(source, separator=separator, **kwds)
+        return self._lazyframe.from_native(native, context=self)
+
+    def read_parquet(self, source: NormalizedPath, **kwds: Any) -> PolarsDataFrame:
+        return self._dataframe.from_native(pl.read_parquet(source, **kwds), context=self)
+
+    def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> PolarsLazyFrame:
+        return self._lazyframe.from_native(pl.scan_parquet(source, **kwds), context=self)
+
     @requires.backend_version(
         (1, 0, 0), "Please use `col` for columns selection instead."
     )