diff --git a/docs/extending.md b/docs/extending.md index 0ddb1e39f7..eef38b1a80 100644 --- a/docs/extending.md +++ b/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). diff --git a/packages/test-plugin/src/test_plugin/namespace.py b/packages/test-plugin/src/test_plugin/namespace.py index 6df24eb2d7..46419516a6 100644 --- a/packages/test-plugin/src/test_plugin/namespace.py +++ b/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() diff --git a/src/narwhals/_arrow/namespace.py b/src/narwhals/_arrow/namespace.py index 744df6f165..e2593cf2bc 100644 --- a/src/narwhals/_arrow/namespace.py +++ b/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]: diff --git a/src/narwhals/_compliant/namespace.py b/src/narwhals/_compliant/namespace.py index eae06cf7b0..1938988e0d 100644 --- a/src/narwhals/_compliant/namespace.py +++ b/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, diff --git a/src/narwhals/_dask/namespace.py b/src/narwhals/_dask/namespace.py index 8384797233..0b071f128e 100644 --- a/src/narwhals/_dask/namespace.py +++ b/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__}" diff --git a/src/narwhals/_duckdb/namespace.py b/src/narwhals/_duckdb/namespace.py index afe67f4901..7247864eef 100644 --- a/src/narwhals/_duckdb/namespace.py +++ b/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) diff --git a/src/narwhals/_ibis/namespace.py b/src/narwhals/_ibis/namespace.py index b6e58fead0..4fb66cc438 100644 --- a/src/narwhals/_ibis/namespace.py +++ b/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) diff --git a/src/narwhals/_pandas_like/namespace.py b/src/narwhals/_pandas_like/namespace.py index b828c16496..52c528a6ef 100644 --- a/src/narwhals/_pandas_like/namespace.py +++ b/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)) diff --git a/src/narwhals/_polars/namespace.py b/src/narwhals/_polars/namespace.py index c552cfeb90..03de33060c 100644 --- a/src/narwhals/_polars/namespace.py +++ b/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." ) diff --git a/src/narwhals/_spark_like/namespace.py b/src/narwhals/_spark_like/namespace.py index 0aff47f79f..5adc059302 100644 --- a/src/narwhals/_spark_like/namespace.py +++ b/src/narwhals/_spark_like/namespace.py @@ -3,7 +3,7 @@ import operator from functools import reduce from itertools import chain -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from narwhals._expression_parsing import ( combine_alias_output_names, @@ -20,6 +20,7 @@ true_divide, ) from narwhals._sql.namespace import SQLNamespace +from narwhals._utils import validate_separators if TYPE_CHECKING: from collections.abc import Callable, Iterable, Mapping @@ -28,8 +29,15 @@ from narwhals._compliant.window import WindowInputs from narwhals._spark_like.dataframe import SQLFrameDataFrame # noqa: F401 + from narwhals._spark_like.utils import SparkReader, SparkSession from narwhals._utils import Implementation, Version - from narwhals.typing import ConcatMethod, CorrelationMethod, IntoDType, PythonLiteral + from narwhals.typing import ( + ConcatMethod, + CorrelationMethod, + IntoDType, + NormalizedPath, + PythonLiteral, + ) # Adjust slight SQL vs PySpark differences FUNCTION_REMAPPINGS = { @@ -68,6 +76,34 @@ def _expr(self) -> type[SparkLikeExpr]: def _lazyframe(self) -> type[SparkLikeLazyFrame]: return SparkLikeLazyFrame + def _session_reader(self, fmt: str, kwds: dict[str, Any]) -> SparkReader: + if (session := kwds.pop("session", None)) is None: + msg = "Spark like backends require a session object to be passed in `kwargs`." + raise ValueError(msg) + + return cast("SparkSession", session).read.format(fmt) + + def scan_csv( + self, source: NormalizedPath, *, separator: str = ",", **kwds: Any + ) -> SparkLikeLazyFrame: + validate_separators(separator, ("sep", "delimiter"), kwds) + reader = self._session_reader("csv", kwds) + native = ( + reader.load(source, sep=separator) + if self._implementation.is_sqlframe() and self._backend_version < (3, 27) + else reader.options(sep=separator, **kwds).load(source) + ) + return self._lazyframe.from_native(native, context=self) + + def scan_parquet(self, source: NormalizedPath, **kwds: Any) -> SparkLikeLazyFrame: + reader = self._session_reader("parquet", kwds) + native = ( + reader.load(source) + if self._implementation.is_sqlframe() and self._backend_version < (3, 27) + else reader.options(**kwds).load(source) + ) + return self._lazyframe.from_native(native, context=self) + @property def _F(self): # type: ignore[no-untyped-def] # noqa: ANN202 if TYPE_CHECKING: diff --git a/src/narwhals/_spark_like/utils.py b/src/narwhals/_spark_like/utils.py index 3de3eef4f2..35c374e755 100644 --- a/src/narwhals/_spark_like/utils.py +++ b/src/narwhals/_spark_like/utils.py @@ -17,16 +17,18 @@ import sqlframe.base.types as sqlframe_types from sqlframe.base.column import Column + from sqlframe.base.readerwriter import _BaseDataFrameReader from sqlframe.base.session import _BaseSession as Session from narwhals._compliant.typing import CompliantLazyFrameAny - from narwhals._spark_like.dataframe import SparkLikeLazyFrame + from narwhals._spark_like.dataframe import SparkLikeLazyFrame, SQLFrameDataFrame from narwhals._spark_like.expr import SparkLikeExpr from narwhals.dtypes import DType from narwhals.typing import IntoDType _NativeDType: TypeAlias = sqlframe_types.DataType SparkSession = Session[Any, Any, Any, Any, Any, Any, Any] + SparkReader = _BaseDataFrameReader[SparkSession, SQLFrameDataFrame, Any] UNITS_DICT = { "y": "year", diff --git a/src/narwhals/_utils.py b/src/narwhals/_utils.py index 30eefdf7f2..90e893e2e8 100644 --- a/src/narwhals/_utils.py +++ b/src/narwhals/_utils.py @@ -129,6 +129,7 @@ IntoSeriesT, MultiIndexSelector, NestedLiteral, + NormalizedPath, SingleIndexSelector, SizedMultiBoolSelector, SizedMultiIndexSelector, @@ -2142,10 +2143,25 @@ def to_pyarrow_table(tbl: pa.Table | pa.RecordBatchReader) -> pa.Table: return tbl +def validate_separators( + separator: str, native_separators: tuple[str, ...], kwds: Mapping[str, Any], / +) -> None: + """Ensure `separator` does not conflict with backend-native aliases passed via `kwds`.""" + for native_separator in native_separators: + if native_separator in kwds and kwds[native_separator] != separator: + msg = ( + f"`separator` and `{native_separator}` do not match: " + f"`separator`={separator} and `{native_separator}`={kwds[native_separator]}." + ) + raise TypeError(msg) + + if sys.platform != "win32": - def normalize_path(source: FileSource, /) -> str: - return source if isinstance(source, str) else str(Path(source)) + def normalize_path(source: FileSource, /) -> NormalizedPath: + from narwhals.typing import NormalizedPath + + return NormalizedPath(source if isinstance(source, str) else str(Path(source))) else: # pragma: no cover # NOTE: On Windows, we need to ensure strings paths do not produce escape sequences. # This module is an example of the issue: @@ -2153,8 +2169,10 @@ def normalize_path(source: FileSource, /) -> str: # If we stringify that, we get: # `'\\narwhals\\narwhals\\_utils.py'` # Which contains 2x `"\n"` characters - def normalize_path(source: FileSource, /) -> str: - return Path(source).as_posix() + def normalize_path(source: FileSource, /) -> NormalizedPath: + from narwhals.typing import NormalizedPath + + return NormalizedPath(Path(source).as_posix()) def extend_bool( diff --git a/src/narwhals/functions.py b/src/narwhals/functions.py index a576c6b33d..fb661554b3 100644 --- a/src/narwhals/functions.py +++ b/src/narwhals/functions.py @@ -622,33 +622,6 @@ def show_versions() -> None: print(f"{k:>13}: {stat}") # noqa: T201 -def _validate_separators( - separator: str, native_separators: tuple[str, ...], **kwargs: Any -) -> None: - for native_separator in native_separators: - if native_separator in kwargs and kwargs[native_separator] != separator: - msg = ( - f"`separator` and `{native_separator}` do not match: " - f"`separator`={separator} and `{native_separator}`={kwargs[native_separator]}." - ) - raise TypeError(msg) - - -def _validate_separator_pyarrow(separator: str, **kwargs: Any) -> Any: - if "parse_options" in kwargs: - parse_options = kwargs.pop("parse_options") - if parse_options.delimiter != separator: - msg = ( - "`separator` and `parse_options.delimiter` do not match: " - f"`separator`={separator} and `delimiter`={parse_options.delimiter}." - ) - raise TypeError(msg) - return kwargs - from pyarrow import csv # ignore-banned-import - - return {"parse_options": csv.ParseOptions(delimiter=separator)} - - def read_csv( source: FileSource, *, @@ -684,44 +657,27 @@ def read_csv( └──────────────────┘ """ impl = Implementation.from_backend(backend) - native_namespace = impl.to_native_namespace() - native_frame: NativeDataFrame - if impl in {Implementation.PANDAS, Implementation.MODIN, Implementation.CUDF}: - _validate_separators(separator, ("sep",), **kwargs) - native_frame = native_namespace.read_csv( - normalize_path(source), sep=separator, **kwargs - ) - elif impl is Implementation.POLARS: - native_frame = native_namespace.read_csv( - normalize_path(source), separator=separator, **kwargs - ) - elif impl is Implementation.PYARROW: - kwargs = _validate_separator_pyarrow(separator, **kwargs) - from pyarrow import csv # ignore-banned-import - - native_frame = csv.read_csv(source, **kwargs) - elif impl in { - Implementation.PYSPARK, - Implementation.DASK, - Implementation.DUCKDB, - Implementation.IBIS, - Implementation.SQLFRAME, - Implementation.PYSPARK_CONNECT, - }: - msg = ( - f"Expected eager backend, found {impl}.\n\n" - f"Hint: use nw.scan_csv(source={source}, backend={backend})" - ) - raise ValueError(msg) - else: # pragma: no cover + if is_eager_allowed(impl): + ns = Version.MAIN.namespace.from_backend(impl).compliant + frame = ns.read_csv(normalize_path(source), separator=separator, **kwargs) + return frame.to_narwhals() + if impl is Implementation.UNKNOWN: # pragma: no cover + native_namespace = impl.to_native_namespace() try: # implementation is UNKNOWN, Narwhals extension using this feature should # implement `read_csv` function in the top-level namespace. - native_frame = native_namespace.read_csv(source=source, **kwargs) + native_frame: NativeDataFrame = native_namespace.read_csv( + source=source, **kwargs + ) except AttributeError as e: msg = "Unknown namespace is expected to implement `read_csv` function." raise AttributeError(msg) from e - return from_native(native_frame, eager_only=True) + return from_native(native_frame, eager_only=True) + msg = ( + f"Expected eager backend, found {impl}.\n\n" + f"Hint: use nw.scan_csv(source={source}, backend={backend})" + ) + raise ValueError(msg) def scan_csv( @@ -764,52 +720,23 @@ def scan_csv( │ z │ 3 │ └─────────┴───────┘ """ - implementation = Implementation.from_backend(backend) - native_namespace = implementation.to_native_namespace() - native_frame: NativeDataFrame | NativeLazyFrame - source = normalize_path(source) - if implementation is Implementation.POLARS: - native_frame = native_namespace.scan_csv(source, separator=separator, **kwargs) - elif implementation in { - Implementation.PANDAS, - Implementation.MODIN, - Implementation.CUDF, - Implementation.DASK, - Implementation.IBIS, - }: - _validate_separators(separator, ("sep",), **kwargs) - native_frame = native_namespace.read_csv(source, sep=separator, **kwargs) - elif implementation is Implementation.DUCKDB: - _validate_separators(separator, ("delimiter", "delim", "sep"), **kwargs) - native_frame = native_namespace.read_csv(source, delimiter=separator, **kwargs) - elif implementation is Implementation.PYARROW: - kwargs = _validate_separator_pyarrow(separator, **kwargs) - from pyarrow import csv # ignore-banned-import - - native_frame = csv.read_csv(source, **kwargs) - elif implementation.is_spark_like(): - _validate_separators(separator, ("sep", "delimiter"), **kwargs) - if (session := kwargs.pop("session", None)) is None: - msg = "Spark like backends require a session object to be passed in `kwargs`." - raise ValueError(msg) - csv_reader = session.read.format("csv") - native_frame = ( - csv_reader.load(source, sep=separator) - if ( - implementation is Implementation.SQLFRAME - and implementation._backend_version() < (3, 27, 0) - ) - else csv_reader.options(sep=separator, **kwargs).load(source) - ) - else: # pragma: no cover + impl = Implementation.from_backend(backend) + if impl is Implementation.UNKNOWN: # pragma: no cover + native_namespace = impl.to_native_namespace() try: # implementation is UNKNOWN, Narwhals extension using this feature should # implement `scan_csv` function in the top-level namespace. - native_frame = native_namespace.scan_csv(source=source, **kwargs) + native_frame: NativeDataFrame | NativeLazyFrame = native_namespace.scan_csv( + source=source, **kwargs + ) except AttributeError as e: msg = "Unknown namespace is expected to implement `scan_csv` function." raise AttributeError(msg) from e - return from_native(native_frame).lazy() + return from_native(native_frame).lazy() + ns = Version.MAIN.namespace.from_backend(impl).compliant + frame = ns.scan_csv(normalize_path(source), separator=separator, **kwargs) + result: LazyFrame[Any] = frame.to_narwhals().lazy() + return result def read_parquet( @@ -847,42 +774,27 @@ def read_parquet( └──────────────────┘ """ impl = Implementation.from_backend(backend) - native_namespace = impl.to_native_namespace() - native_frame: NativeDataFrame - if impl in { - Implementation.POLARS, - Implementation.PANDAS, - Implementation.MODIN, - Implementation.CUDF, - }: - source = normalize_path(source) - native_frame = native_namespace.read_parquet(source, **kwargs) - elif impl is Implementation.PYARROW: - import pyarrow.parquet as pq # ignore-banned-import - - native_frame = pq.read_table(source, **kwargs) # type: ignore[arg-type] - elif impl in { - Implementation.PYSPARK, - Implementation.DASK, - Implementation.DUCKDB, - Implementation.IBIS, - Implementation.SQLFRAME, - Implementation.PYSPARK_CONNECT, - }: - msg = ( - f"Expected eager backend, found {impl}.\n\n" - f"Hint: use nw.scan_parquet(source={source}, backend={backend})" - ) - raise ValueError(msg) - else: # pragma: no cover + if is_eager_allowed(impl): + ns = Version.MAIN.namespace.from_backend(impl).compliant + frame = ns.read_parquet(normalize_path(source), **kwargs) + return frame.to_narwhals() + if impl is Implementation.UNKNOWN: # pragma: no cover + native_namespace = impl.to_native_namespace() try: # implementation is UNKNOWN, Narwhals extension using this feature should # implement `read_parquet` function in the top-level namespace. - native_frame = native_namespace.read_parquet(source=source, **kwargs) + native_frame: NativeDataFrame = native_namespace.read_parquet( + source=source, **kwargs + ) except AttributeError as e: msg = "Unknown namespace is expected to implement `read_parquet` function." raise AttributeError(msg) from e - return from_native(native_frame, eager_only=True) + return from_native(native_frame, eager_only=True) + msg = ( + f"Expected eager backend, found {impl}.\n\n" + f"Hint: use nw.scan_parquet(source={source}, backend={backend})" + ) + raise ValueError(msg) def scan_parquet( @@ -947,48 +859,23 @@ def scan_parquet( | b: [[4,5]] | └──────────────────┘ """ - implementation = Implementation.from_backend(backend) - native_namespace = implementation.to_native_namespace() - native_frame: NativeDataFrame | NativeLazyFrame - source = normalize_path(source) - if implementation is Implementation.POLARS: - native_frame = native_namespace.scan_parquet(source, **kwargs) - elif implementation in { - Implementation.PANDAS, - Implementation.MODIN, - Implementation.CUDF, - Implementation.DASK, - Implementation.DUCKDB, - Implementation.IBIS, - }: - native_frame = native_namespace.read_parquet(source, **kwargs) - elif implementation is Implementation.PYARROW: - import pyarrow.parquet as pq # ignore-banned-import - - native_frame = pq.read_table(source, **kwargs) - elif implementation.is_spark_like(): - if (session := kwargs.pop("session", None)) is None: - msg = "Spark like backends require a session object to be passed in `kwargs`." - raise ValueError(msg) - pq_reader = session.read.format("parquet") - native_frame = ( - pq_reader.load(source) - if ( - implementation is Implementation.SQLFRAME - and implementation._backend_version() < (3, 27, 0) - ) - else pq_reader.options(**kwargs).load(source) - ) - - else: # pragma: no cover + impl = Implementation.from_backend(backend) + if impl is Implementation.UNKNOWN: # pragma: no cover + native_namespace = impl.to_native_namespace() try: # implementation is UNKNOWN, Narwhals extension using this feature should # implement `scan_parquet` function in the top-level namespace. - native_frame = native_namespace.scan_parquet(source=source, **kwargs) + native_frame: NativeDataFrame | NativeLazyFrame = ( + native_namespace.scan_parquet(source=source, **kwargs) + ) except AttributeError as e: msg = "Unknown namespace is expected to implement `scan_parquet` function." raise AttributeError(msg) from e - return from_native(native_frame).lazy() + return from_native(native_frame).lazy() + ns = Version.MAIN.namespace.from_backend(impl).compliant + frame = ns.scan_parquet(normalize_path(source), **kwargs) + result: LazyFrame[Any] = frame.to_narwhals().lazy() + return result def col(*names: str | Iterable[str]) -> Expr: diff --git a/src/narwhals/typing.py b/src/narwhals/typing.py index 9a9f93d833..9173c0b612 100644 --- a/src/narwhals/typing.py +++ b/src/narwhals/typing.py @@ -1,7 +1,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeVar, Union +from typing import TYPE_CHECKING, Any, Literal, NewType, Protocol, TypeVar, Union from narwhals._compliant import CompliantDataFrame, CompliantLazyFrame, CompliantSeries from narwhals._native import ( @@ -374,6 +374,13 @@ def Binary(self) -> type[dtypes.Binary]: ... [`pathlib.Path`]: https://docs.python.org/3/library/pathlib.html#pathlib.Path """ +NormalizedPath = NewType("NormalizedPath", str) +"""A [`FileSource`][narwhals.typing.FileSource] normalized via `narwhals._utils.normalize_path`. + +The compliant-namespace IO methods (`read_csv`, `scan_csv`, `read_parquet`, `scan_parquet`) +take an already-normalized path and forward `kwds` to the native reader. +""" + # Annotations for `__getitem__` methods _T = TypeVar("_T")