polars_pylance

polars_pylance

Lazy, streaming Lance <-> Polars integration.

Reading gives a real LazyFrame: the optimizer pushes column projections, filters and row limits into Lance, and batches are pulled only as the streaming engine consumes them. Writing streams a query into Lance batch by batch, so neither direction holds the dataset (or a whole fragment) in RAM.

>>> import polars as pl
>>> import polars_pylance as pll
>>> lf = pll.scan_lance("data.lance")  # doctest: +SKIP
>>> pll.sink_lance(
...     lf.filter(pl.col("score") > 0.9), "filtered.lance"
... )  # doctest: +SKIP

LanceScanOptions dataclass

Per-scan Lance reader tuning. Immutable and picklable.

Every field maps to the identically named argument of lance.LanceDataset.scanner. None means "leave it to Lance".

Parameters:

Name	Type	Description	Default
batch_size	int | None	Rows per record batch handed to Polars.	25000
batch_readahead	int | None	Batches decoded ahead of the consumer, per fragment.	1
fragment_readahead	int | None	Fragments read concurrently.	1
io_buffer_size	int | None	Size of the Lance IO buffer, in bytes. This is the single biggest lever on peak memory; Lance's own default is 2 GiB.	32 * MIB
scan_in_order	bool | None	Yield fragments in order. Out-of-order scanning is faster but lets more data accumulate in flight.	True
use_scalar_index	bool | None	Whether pushed-down predicates may use scalar indices.	None
late_materialization	bool | list[str] | None	Defer loading of large columns until after filtering. Either a bool for all columns or a list of column names.	None

Source code in src/polars_pylance/_options.py

@dataclass(frozen=True)
class LanceScanOptions:
    """Per-scan Lance reader tuning. Immutable and picklable.

    Every field maps to the identically named argument of
    `lance.LanceDataset.scanner`. `None` means "leave it to Lance".

    Args:
        batch_size: Rows per record batch handed to Polars.
        batch_readahead: Batches decoded ahead of the consumer, per fragment.
        fragment_readahead: Fragments read concurrently.
        io_buffer_size: Size of the Lance IO buffer, in bytes. This is the single
            biggest lever on peak memory; Lance's own default is 2 GiB.
        scan_in_order: Yield fragments in order. Out-of-order scanning is faster but
            lets more data accumulate in flight.
        use_scalar_index: Whether pushed-down predicates may use scalar indices.
        late_materialization: Defer loading of large columns until after filtering.
            Either a bool for all columns or a list of column names.
    """

    batch_size: int | None = 25_000
    batch_readahead: int | None = 1
    fragment_readahead: int | None = 1
    io_buffer_size: int | None = 32 * MIB
    scan_in_order: bool | None = True
    use_scalar_index: bool | None = None
    late_materialization: bool | list[str] | None = None

    @classmethod
    def throughput(cls, **overrides: Any) -> LanceScanOptions:  # noqa: ANN401
        """Restore Lance's own aggressive read-ahead, for when RAM is plentiful.

        Roughly 1.6x the peak memory of the defaults on a large-payload scan, in
        exchange for more IO parallelism.
        """
        return cls(
            batch_size=None,
            batch_readahead=None,
            fragment_readahead=None,
            io_buffer_size=None,
            scan_in_order=None,
            **overrides,
        )

    def replace(self, **overrides: Any) -> LanceScanOptions:  # noqa: ANN401
        """Return a copy with `overrides` applied."""
        current = {f.name: getattr(self, f.name) for f in fields(self)}
        unknown = set(overrides) - set(current)
        if unknown:
            msg = f"unknown LanceScanOptions fields: {sorted(unknown)}"
            raise TypeError(msg)
        return type(self)(**{**current, **overrides})

    def to_scan_kwargs(self) -> dict[str, Any]:
        """Render as `scanner()` keyword arguments, omitting unset fields."""
        return {
            f.name: value
            for f in fields(self)
            if (value := getattr(self, f.name)) is not None
        }

throughput classmethod

throughput(**overrides: Any) -> LanceScanOptions

Restore Lance's own aggressive read-ahead, for when RAM is plentiful.

Roughly 1.6x the peak memory of the defaults on a large-payload scan, in exchange for more IO parallelism.

Source code in src/polars_pylance/_options.py

@classmethod
def throughput(cls, **overrides: Any) -> LanceScanOptions:  # noqa: ANN401
    """Restore Lance's own aggressive read-ahead, for when RAM is plentiful.

    Roughly 1.6x the peak memory of the defaults on a large-payload scan, in
    exchange for more IO parallelism.
    """
    return cls(
        batch_size=None,
        batch_readahead=None,
        fragment_readahead=None,
        io_buffer_size=None,
        scan_in_order=None,
        **overrides,
    )

replace

replace(**overrides: Any) -> LanceScanOptions

Return a copy with overrides applied.

Source code in src/polars_pylance/_options.py

def replace(self, **overrides: Any) -> LanceScanOptions:  # noqa: ANN401
    """Return a copy with `overrides` applied."""
    current = {f.name: getattr(self, f.name) for f in fields(self)}
    unknown = set(overrides) - set(current)
    if unknown:
        msg = f"unknown LanceScanOptions fields: {sorted(unknown)}"
        raise TypeError(msg)
    return type(self)(**{**current, **overrides})

to_scan_kwargs

to_scan_kwargs() -> dict[str, Any]

Render as scanner() keyword arguments, omitting unset fields.

Source code in src/polars_pylance/_options.py

def to_scan_kwargs(self) -> dict[str, Any]:
    """Render as `scanner()` keyword arguments, omitting unset fields."""
    return {
        f.name: value
        for f in fields(self)
        if (value := getattr(self, f.name)) is not None
    }

LanceFilter dataclass

A Lance SQL filter lowered from a Polars predicate.

Attributes:

Name	Type	Description
sql	str	The filter string, ready for LanceDataset.scanner(filter=...).
exact	bool	True when the filter keeps exactly the rows the predicate keeps. False when part of the predicate was dropped, leaving a superset; the caller must then keep evaluating the predicate.

Source code in src/polars_pylance/_predicate.py

@dataclass(frozen=True)
class LanceFilter:
    """A Lance SQL filter lowered from a Polars predicate.

    Attributes:
        sql: The filter string, ready for `LanceDataset.scanner(filter=...)`.
        exact: True when the filter keeps exactly the rows the predicate keeps. False
            when part of the predicate was dropped, leaving a superset; the caller must
            then keep evaluating the predicate.
    """

    sql: str
    exact: bool

LanceScanSpec dataclass

Everything needed to reproduce a scan, and nothing that cannot be pickled.

Holding a URI rather than an open lance.LanceDataset is what lets a scan be serialized into a query plan and executed elsewhere, which is the prerequisite for Polars Cloud.

The fields are the arguments of scan_lance, which documents them, with two differences: prefilter has already been lowered to a Lance SQL string, and fragment_ids is the resolved list that fragments selected.

Source code in src/polars_pylance/_scan.py

@dataclass
class LanceScanSpec:
    """Everything needed to reproduce a scan, and nothing that cannot be pickled.

    Holding a URI rather than an open `lance.LanceDataset` is what lets a
    scan be serialized into a query plan and executed elsewhere, which is the
    prerequisite for Polars Cloud.

    The fields are the arguments of [`scan_lance`][polars_pylance.scan_lance], which
    documents them, with two differences: `prefilter` has already been lowered
    to a Lance SQL string, and `fragment_ids` is the resolved list that
    `fragments` selected.
    """

    uri: str
    version: int | str | None = None
    storage_options: dict[str, str] | None = None
    options: LanceScanOptions = field(default_factory=LanceScanOptions)
    nearest: dict[str, Any] | None = None
    full_text_query: str | dict[str, Any] | None = None
    # Already lowered to Lance SQL by `scan_lance`, so the spec stays a plain
    # picklable record and an unsupported prefilter fails at the call site.
    prefilter: str | None = None
    with_row_id: bool = False
    with_row_address: bool = False
    fragment_ids: list[int] | None = None
    predicate_pushdown: bool = True

    # -- dataset access ----------------------------------------------------

    def open(self) -> lance.LanceDataset:
        return lance.dataset(
            self.uri, version=self.version, storage_options=self.storage_options
        )

    def scanner(
        self,
        dataset: lance.LanceDataset,
        *,
        columns: list[str] | None = None,
        filter: str | None = None,
        limit: int | None = None,
        prefilter: bool = False,
    ) -> lance.LanceScanner:
        kwargs: dict[str, Any] = {
            **self.options.to_scan_kwargs(),
            "columns": columns,
            "filter": filter,
            "limit": limit,
        }
        if prefilter:
            # Lance has one filter slot; this says it restricts the rows the
            # search runs over instead of filtering the search's result.
            kwargs["prefilter"] = True
        if self.nearest is not None:
            kwargs["nearest"] = self.nearest
        if self.full_text_query is not None:
            kwargs["full_text_query"] = self.full_text_query
        if self.with_row_id:
            kwargs["with_row_id"] = True
        if self.with_row_address:
            kwargs["with_row_address"] = True
        if self.fragment_ids is not None:
            by_id = {f.fragment_id: f for f in dataset.get_fragments()}
            missing = [i for i in self.fragment_ids if i not in by_id]
            if missing:
                msg = f"no such fragment(s) in {self.uri}: {missing}"
                raise ValueError(msg)
            kwargs["fragments"] = [by_id[i] for i in self.fragment_ids]
        return dataset.scanner(**kwargs)

    def arrow_schema(self, dataset: lance.LanceDataset | None = None) -> pa.Schema:
        """Full output schema, including any Lance-generated columns."""
        dataset = dataset if dataset is not None else self.open()
        return self.scanner(dataset).projected_schema

    def polars_schema(self, dataset: lance.LanceDataset | None = None) -> pl.Schema:
        arrow = self.arrow_schema(dataset)
        empty = pl.from_arrow(arrow.empty_table())
        assert isinstance(empty, pl.DataFrame)
        return empty.schema

    # -- batch production --------------------------------------------------

    def iter_frames(
        self,
        dataset: lance.LanceDataset,
        *,
        projection: Sequence[str] | None = None,
        filter: str | None = None,
        limit: int | None = None,
        prefilter: bool = False,
    ) -> Iterator[pl.DataFrame]:
        """Stream `projection` out of Lance as Polars frames.

        Lazy by construction: nothing is read until the consumer pulls, and
        dropping the generator early stops the scan.
        """
        columns = None if projection is None else self._physical_columns(projection)
        scanner = self.scanner(
            dataset, columns=columns, filter=filter, limit=limit, prefilter=prefilter
        )

        remaining = limit
        # A pushed-down predicate may be dropped and the scan retried without
        # it, because Polars can finish it. A prefilter has no such second
        # chance: dropping it would silently widen what the search ranked.
        for batch in self._batches(
            scanner, fallback=filter is not None and not prefilter
        ):
            if batch.num_rows == 0:
                continue
            if remaining is not None:
                if remaining <= 0:
                    break
                if batch.num_rows > remaining:
                    batch = batch.slice(0, remaining)  # noqa: PLW2901
                remaining -= batch.num_rows

            if batch.num_columns == 0:
                # A column-less batch still carries a row count (this is what a
                # bare `pl.len()` projects to), but Arrow -> Polars conversion
                # would lose it.
                yield pl.DataFrame(height=batch.num_rows)
                continue

            frame = pl.from_arrow(batch)
            assert isinstance(frame, pl.DataFrame)
            if projection is not None and frame.columns != list(projection):
                # Lance appends generated columns after the requested ones; the
                # engine expects exactly the projection, in order.
                frame = frame.select(projection)
            yield frame

    @staticmethod
    def _batches(
        scanner: lance.LanceScanner, *, fallback: bool
    ) -> Iterator[pa.RecordBatch]:
        """`scanner.to_batches()`, reporting a rejected filter as such.

        Lance validates a filter while planning, on the first pull, so telling
        that failure from any other lets the caller retry without it. Only when
        `fallback` says a retry would still give the same rows; otherwise the
        error is Lance's own and is left to reach the caller.
        """
        iterator = iter(scanner.to_batches())
        try:
            first = next(iterator, None)
        except Exception as exc:
            if not fallback:
                raise
            raise _FilterRejected(str(exc)) from exc
        if first is None:
            return
        yield first
        yield from iterator

    def _physical_columns(self, projection: Sequence[str]) -> list[str]:
        # Generated columns are added by the scanner itself and must not appear
        # in `columns=`; an empty result is legal and means "generated only".
        return [c for c in projection if c not in VIRTUAL_COLUMNS]

arrow_schema

arrow_schema(dataset: LanceDataset | None = None) -> Schema

Full output schema, including any Lance-generated columns.

Source code in src/polars_pylance/_scan.py

def arrow_schema(self, dataset: lance.LanceDataset | None = None) -> pa.Schema:
    """Full output schema, including any Lance-generated columns."""
    dataset = dataset if dataset is not None else self.open()
    return self.scanner(dataset).projected_schema

iter_frames

iter_frames(dataset: LanceDataset, *, projection: Sequence[str] | None = None, filter: str | None = None, limit: int | None = None, prefilter: bool = False) -> Iterator[DataFrame]

Stream projection out of Lance as Polars frames.

Lazy by construction: nothing is read until the consumer pulls, and dropping the generator early stops the scan.

Source code in src/polars_pylance/_scan.py

def iter_frames(
    self,
    dataset: lance.LanceDataset,
    *,
    projection: Sequence[str] | None = None,
    filter: str | None = None,
    limit: int | None = None,
    prefilter: bool = False,
) -> Iterator[pl.DataFrame]:
    """Stream `projection` out of Lance as Polars frames.

    Lazy by construction: nothing is read until the consumer pulls, and
    dropping the generator early stops the scan.
    """
    columns = None if projection is None else self._physical_columns(projection)
    scanner = self.scanner(
        dataset, columns=columns, filter=filter, limit=limit, prefilter=prefilter
    )

    remaining = limit
    # A pushed-down predicate may be dropped and the scan retried without
    # it, because Polars can finish it. A prefilter has no such second
    # chance: dropping it would silently widen what the search ranked.
    for batch in self._batches(
        scanner, fallback=filter is not None and not prefilter
    ):
        if batch.num_rows == 0:
            continue
        if remaining is not None:
            if remaining <= 0:
                break
            if batch.num_rows > remaining:
                batch = batch.slice(0, remaining)  # noqa: PLW2901
            remaining -= batch.num_rows

        if batch.num_columns == 0:
            # A column-less batch still carries a row count (this is what a
            # bare `pl.len()` projects to), but Arrow -> Polars conversion
            # would lose it.
            yield pl.DataFrame(height=batch.num_rows)
            continue

        frame = pl.from_arrow(batch)
        assert isinstance(frame, pl.DataFrame)
        if projection is not None and frame.columns != list(projection):
            # Lance appends generated columns after the requested ones; the
            # engine expects exactly the projection, in order.
            frame = frame.select(projection)
        yield frame

to_lance_filter

to_lance_filter(predicate: Expr, *, max_in_list: int = MAX_IN_LIST, schema: Schema | None = None) -> LanceFilter | None

Lower predicate to a Lance SQL filter, or None if nothing can be pushed.

Parameters:

Name	Type	Description	Default
predicate	Expr	Any boolean Polars expression, however deeply nested.	required
max_in_list	int	Largest is_in membership list to spell out as SQL IN.	MAX_IN_LIST
schema	Schema | None	The scanned schema, when the caller has it. Used to drop a promotion the schema shows is a no-op, since a CAST around an indexed column costs its scalar index, and to tell a string + from an arithmetic one.	None

Examples:

>>> import polars as pl
>>> from polars_pylance import to_lance_filter
>>> to_lance_filter(pl.col("cat").str.starts_with("b"))
LanceFilter(sql="starts_with(`cat`, 'b')", exact=True)
>>> to_lance_filter(
...     pl.col("cat").str.extract(r"(\d+)").is_null() & (pl.col("id") > 3)
... )
LanceFilter(sql='(`id` > 3)', exact=False)
>>> to_lance_filter(pl.col("id").hash() > 3) is None
True

Source code in src/polars_pylance/_predicate.py

def to_lance_filter(
    predicate: pl.Expr,
    *,
    max_in_list: int = MAX_IN_LIST,
    schema: pl.Schema | None = None,
) -> LanceFilter | None:
    r"""Lower `predicate` to a Lance SQL filter, or None if nothing can be pushed.

    Args:
        predicate: Any boolean Polars expression, however deeply nested.
        max_in_list: Largest `is_in` membership list to spell out as SQL `IN`.
        schema: The scanned schema, when the caller has it. Used to drop a promotion the
            schema shows is a no-op, since a `CAST` around an indexed column costs its
            scalar index, and to tell a string `+` from an arithmetic one.

    Examples:
        >>> import polars as pl
        >>> from polars_pylance import to_lance_filter
        >>> to_lance_filter(pl.col("cat").str.starts_with("b"))
        LanceFilter(sql="starts_with(`cat`, 'b')", exact=True)
        >>> to_lance_filter(
        ...     pl.col("cat").str.extract(r"(\d+)").is_null() & (pl.col("id") > 3)
        ... )
        LanceFilter(sql='(`id` > 3)', exact=False)
        >>> to_lance_filter(pl.col("id").hash() > 3) is None
        True
    """
    try:
        tree = json.loads(predicate.meta.serialize(format="json"))
    except Exception:  # noqa: BLE001 - see below
        # No tree means nothing to lower, and pushdown is optional, so this
        # declines rather than reaching the caller. The failure this is known to
        # catch is a `ComputeError` from a UDF closing over something
        # unpicklable; the family polars raises here is not documented, and a
        # wrong guess would turn a missed optimization into a failed query. A
        # plain UDF does serialize: it is declined by the walk, as an
        # `AnonymousFunction` node it has no spelling for.
        return None

    lowering = _Lowering(max_in_list=max_in_list, schema=schema)
    try:
        sql, exact = lowering.predicate(tree)
    except (_Decline, RecursionError):
        return None
    if sql is None:
        return None
    return LanceFilter(sql=sql, exact=exact)

scan_lance

scan_lance(source: str | Path | LanceDataset, *, version: int | str | None = None, storage_options: dict[str, str] | None = None, options: LanceScanOptions | None = None, nearest: dict[str, Any] | None = None, full_text_query: str | dict[str, Any] | None = None, prefilter: str | Expr | None = None, with_row_id: bool = False, with_row_address: bool = False, fragments: Sequence[int] | None = None, predicate_pushdown: bool = True) -> LazyFrame

Lazily read a Lance dataset as a Polars LazyFrame.

Nothing is read when this returns. Column projections, filters and row limits are pushed into Lance; batches are pulled only as the query consumes them, so .head() stops the scan early. Use engine="streaming" when collecting: the in-memory engine materialises the whole result and gives up the memory advantage.

The filter becomes a Lance SQL filter, so is_in, string functions, arithmetic, temporal parts and list or struct access reach the scanner, none of which a PyArrow expression can carry. A predicate that only partly translates is pushed as far as it goes and finished in Polars. to_lance_filter shows what a given one lowers to.

Parameters:

Name	Type	Description	Default
source	str | Path | LanceDataset	Dataset URI, path, or an open lance.LanceDataset. Passing a dataset object pins the scan to that dataset's version; passing a URI reads whatever version is current when the query runs.	required
version	int | str | None	Read a specific version (or tag) instead of the latest.	None
storage_options	dict[str, str] | None	Object-store credentials and settings, passed to Lance.	None
options	LanceScanOptions | None	Reader tuning; see LanceScanOptions. The defaults favour bounded memory over raw IO parallelism.	None
nearest	dict[str, Any] | None	Lance vector-search specification, e.g. {"column": "vector", "q": query, "k": 10}. Adds a _distance column. Not expressible as a Polars predicate, hence a scan argument.	None
full_text_query	str | dict[str, Any] | None	Lance full-text search query. Adds a _score column.	None
prefilter	str | Expr | None	Restrict which rows nearest or full_text_query may return before the search runs, as Lance SQL or a Polars expression. This is not the same question as .filter() on the result, which ranks first and filters after and so may return fewer than k rows; with a prefilter the search picks its k from the surviving rows only. A Polars expression that does not translate exactly is an error rather than a postfilter, since nothing downstream can repair a candidate set the search has already used.	None
with_row_id	bool	Include Lance's stable _rowid column.	False
with_row_address	bool	Include Lance's physical _rowaddr column.	False
fragments	Sequence[int] | None	Restrict the scan to these fragment ids. See scan_lance_fragments for the sharded form.	None
predicate_pushdown	bool	Set to False to keep filtering entirely in Polars. Worth trying if you depend on Polars' null comparison semantics, which differ from SQL's.	True

Examples:

>>> lf = scan_lance("s3://bucket/data.lance")
>>> lf.filter(pl.col("label").is_in([3, 7])).select("id", "score").collect(
...     engine="streaming"
... )

Source code in src/polars_pylance/_scan.py

def scan_lance(
    source: str | Path | lance.LanceDataset,
    *,
    version: int | str | None = None,
    storage_options: dict[str, str] | None = None,
    options: LanceScanOptions | None = None,
    nearest: dict[str, Any] | None = None,
    full_text_query: str | dict[str, Any] | None = None,
    prefilter: str | pl.Expr | None = None,
    with_row_id: bool = False,
    with_row_address: bool = False,
    fragments: Sequence[int] | None = None,
    predicate_pushdown: bool = True,
) -> pl.LazyFrame:
    """Lazily read a Lance dataset as a Polars `LazyFrame`.

    Nothing is read when this returns. Column projections, filters and row
    limits are pushed into Lance; batches are pulled only as the query consumes
    them, so `.head()` stops the scan early. Use `engine="streaming"` when
    collecting: the in-memory engine materialises the whole result and gives up
    the memory advantage.

    The filter becomes a Lance SQL filter, so `is_in`, string functions, arithmetic,
    temporal parts and list or struct access reach the scanner, none of which a PyArrow
    expression can carry. A predicate that only partly translates is pushed as far as it
    goes and finished in Polars. [`to_lance_filter`][polars_pylance.to_lance_filter]
    shows what a given one lowers to.

    Args:
        source: Dataset URI, path, or an open `lance.LanceDataset`. Passing a dataset
            object pins the scan to that dataset's version; passing a URI reads whatever
            version is current when the query runs.
        version: Read a specific version (or tag) instead of the latest.
        storage_options: Object-store credentials and settings, passed to Lance.
        options: Reader tuning; see
            [`LanceScanOptions`][polars_pylance.LanceScanOptions]. The defaults favour
            bounded memory over raw IO parallelism.
        nearest: Lance vector-search specification, e.g. `{"column": "vector", "q":
            query, "k": 10}`. Adds a `_distance` column. Not expressible as a Polars
            predicate, hence a scan argument.
        full_text_query: Lance full-text search query. Adds a `_score` column.
        prefilter: Restrict which rows `nearest` or `full_text_query` may return before
            the search runs, as Lance SQL or a Polars expression. This is not the same
            question as `.filter()` on the result, which ranks first and filters after
            and so may return fewer than `k` rows; with a prefilter the search picks its
            `k` from the surviving rows only. A Polars expression that does not
            translate exactly is an error rather than a postfilter, since nothing
            downstream can repair a candidate set the search has already used.
        with_row_id: Include Lance's stable `_rowid` column.
        with_row_address: Include Lance's physical `_rowaddr` column.
        fragments: Restrict the scan to these fragment ids. See
            [`scan_lance_fragments`][polars_pylance.scan_lance_fragments] for the
            sharded form.
        predicate_pushdown: Set to False to keep filtering entirely in Polars. Worth
            trying if you depend on Polars' null comparison semantics, which differ from
            SQL's.

    Examples:
        >>> lf = scan_lance("s3://bucket/data.lance")  # doctest: +SKIP
        >>> lf.filter(pl.col("label").is_in([3, 7])).select("id", "score").collect(
        ...     engine="streaming"
        ... )  # doctest: +SKIP
    """
    if isinstance(source, lance.LanceDataset):
        uri = source.uri
        version = version if version is not None else source.version
    else:
        uri = str(source)

    spec = LanceScanSpec(
        uri=uri,
        version=version,
        storage_options=storage_options,
        options=options if options is not None else LanceScanOptions(),
        nearest=nearest,
        full_text_query=full_text_query,
        prefilter=None if prefilter is None else _prefilter_sql(prefilter),
        with_row_id=with_row_id,
        with_row_address=with_row_address,
        fragment_ids=list(fragments) if fragments is not None else None,
        predicate_pushdown=predicate_pushdown,
    )

    return _io_plugin_lazyframe(spec)

scan_lance_fragments

scan_lance_fragments(source: str | Path | LanceDataset, *, n_shards: int | None = None, **kwargs: Any) -> list[LazyFrame]

Return one LazyFrame per fragment (or per shard) of a Lance dataset.

Lance fragments are the natural unit of parallelism: each is a self-contained set of files. Use this to fan a scan out over threads, processes or workers, then recombine with polars.concat. Also the manual parallelisation route if a distributed planner refuses a Python scan node.

Parameters:

Name	Type	Description	Default
source	str | Path | LanceDataset	Dataset URI, path, or an open lance.LanceDataset, as scan_lance takes it.	required
n_shards	int | None	Group the fragments into this many LazyFrames instead of returning one each. Fragments are dealt round-robin, so shards stay even when the last fragment is a short one. Asking for more shards than there are fragments yields one fragment per shard, not empty shards.	None
**kwargs	Any	Forwarded to scan_lance for every shard, so version, options, prefilter and the rest apply to all of them alike.	{}

Examples:

>>> shards = scan_lance_fragments("data.lance", n_shards=4)
>>> pl.concat(shards).collect(engine="streaming")

Source code in src/polars_pylance/_scan.py

def scan_lance_fragments(
    source: str | Path | lance.LanceDataset,
    *,
    n_shards: int | None = None,
    **kwargs: Any,  # noqa: ANN401 - forwarded to `scan_lance` as given
) -> list[pl.LazyFrame]:
    """Return one LazyFrame per fragment (or per shard) of a Lance dataset.

    Lance fragments are the natural unit of parallelism: each is a self-contained set of
    files. Use this to fan a scan out over threads, processes or workers, then recombine
    with [`polars.concat`][polars.concat]. Also the manual parallelisation route if a
    distributed planner refuses a Python scan node.

    Args:
        source: Dataset URI, path, or an open `lance.LanceDataset`, as
            [`scan_lance`][polars_pylance.scan_lance] takes it.
        n_shards: Group the fragments into this many LazyFrames instead of returning one
            each. Fragments are dealt round-robin, so shards stay even when the last
            fragment is a short one. Asking for more shards than there are fragments
            yields one fragment per shard, not empty shards.
        **kwargs: Forwarded to [`scan_lance`][polars_pylance.scan_lance] for every
            shard, so `version`, `options`, `prefilter` and the rest apply to all of
            them alike.

    Examples:
        >>> shards = scan_lance_fragments("data.lance", n_shards=4)  # doctest: +SKIP
        >>> pl.concat(shards).collect(engine="streaming")  # doctest: +SKIP
    """
    dataset = (
        source
        if isinstance(source, lance.LanceDataset)
        else lance.dataset(
            str(source),
            version=kwargs.get("version"),
            storage_options=kwargs.get("storage_options"),
        )
    )
    ids = [f.fragment_id for f in dataset.get_fragments()]
    if not ids:
        return [scan_lance(source, fragments=[], **kwargs)]

    if n_shards is None:
        groups = [[i] for i in ids]
    else:
        if n_shards < 1:
            msg = f"n_shards must be >= 1, got {n_shards}"
            raise ValueError(msg)
        groups = [chunk for s in range(n_shards) if (chunk := ids[s::n_shards])]

    return [scan_lance(source, fragments=g, **kwargs) for g in groups]

commit_lance_fragments

commit_lance_fragments(uri: str, fragments: list[Any], *, schema: Schema, mode: Literal['create', 'overwrite', 'append'] = 'create', storage_options: dict[str, str] | None = None) -> LanceDataset

Publish already-written fragments as one dataset version.

The second half of a distributed write: the fragments' data files are on storage but no manifest references them, so nothing has been published yet. This is the single commit that makes them a version, whether they were written by threads (write_lance_fragments) or by Polars Cloud workers (sink_lance_remote).

Parameters:

Name	Type	Description	Default
uri	str	Destination dataset URI or path.	required
fragments	list[Any]	The lance.fragment.FragmentMetadata records returned by whatever wrote the data files.	required
schema	Schema	Schema to install on the new version. Used by "create" and "overwrite"; "append" keeps the existing dataset's schema and field ids instead.	required
mode	Literal['create', 'overwrite', 'append']	"create" (fail if the dataset exists), "overwrite" (replace its contents with these fragments), or "append" (add them to it).	'create'
storage_options	dict[str, str] | None	Object-store credentials and settings, passed to Lance.	None

Returns:

Type	Description
LanceDataset	lance.LanceDataset: The dataset at the version this commit created.

Source code in src/polars_pylance/_sink.py

def commit_lance_fragments(
    uri: str,
    fragments: list[Any],
    *,
    schema: pa.Schema,
    mode: Literal["create", "overwrite", "append"] = "create",
    storage_options: dict[str, str] | None = None,
) -> lance.LanceDataset:
    """Publish already-written fragments as one dataset version.

    The second half of a distributed write: the fragments' data files are on storage but
    no manifest references them, so nothing has been published yet. This is the single
    commit that makes them a version, whether they were written by threads
    ([`write_lance_fragments`][polars_pylance.write_lance_fragments]) or by Polars Cloud
    workers ([`sink_lance_remote`][polars_pylance.cloud.sink_lance_remote]).

    Args:
        uri: Destination dataset URI or path.
        fragments: The `lance.fragment.FragmentMetadata` records returned by whatever
            wrote the data files.
        schema: Schema to install on the new version. Used by `"create"` and
            `"overwrite"`; `"append"` keeps the existing dataset's schema and field ids
            instead.
        mode: `"create"` (fail if the dataset exists), `"overwrite"` (replace its
            contents with these fragments), or `"append"` (add them to it).
        storage_options: Object-store credentials and settings, passed to Lance.

    Returns:
        lance.LanceDataset: The dataset at the version this commit created.
    """
    operation: lance.LanceOperation.BaseOperation
    if mode == "append":
        operation = lance.LanceOperation.Append(fragments)
        read_version = lance.dataset(uri, storage_options=storage_options).version
        return lance.LanceDataset.commit(
            uri, operation, read_version=read_version, storage_options=storage_options
        )

    if mode == "create" and _dataset_exists(uri, storage_options):
        msg = (
            f"dataset already exists at {uri!r}; use mode='overwrite' to replace "
            "its contents or mode='append' to add to it"
        )
        raise FileExistsError(msg)

    operation = lance.LanceOperation.Overwrite(schema, fragments)
    return lance.LanceDataset.commit(uri, operation, storage_options=storage_options)

sink_lance

sink_lance(lf: LazyFrame | DataFrame, target: str | Path | LanceDataset, *, mode: WriteMode = 'create', on: str | list[str] | None = None, chunk_size: int = DEFAULT_CHUNK_SIZE, engine: EngineType = 'streaming', lazy: bool = False, **lance_write_kwargs: Any) -> LanceDataset | LazyFrame

Write a Polars DataFrame or stream a LazyFrame into a Lance dataset.

A LazyFrame query is executed in batches and handed to Lance as it goes, so its result is never materialised in full. A DataFrame is accepted for convenience, but is already materialised before this function is called.

Parameters:

Name	Type	Description	Default
lf	LazyFrame | DataFrame	The eager DataFrame or lazy query to write.	required
target	str | Path | LanceDataset	Destination URI, path, or an existing lance.LanceDataset.	required
mode	WriteMode	"create" (fail if it exists), "append", "overwrite" (new version), or "merge" for an upsert, which requires on.	'create'
on	str | list[str] | None	Join key(s) for mode="merge".	None
chunk_size	int	Rows buffered per batch handed to Lance.	DEFAULT_CHUNK_SIZE
engine	EngineType	Polars engine. Leave at "streaming"; "in-memory" defeats the purpose by materialising the result first.	'streaming'
lazy	bool	Return a LazyFrame that performs the write when collected, instead of writing immediately, so the Lance write can sit inside a larger deferred plan. Collecting it yields a one-row summary of what was written.	False
**lance_write_kwargs	Any	Passed through to lance.write_dataset, e.g. max_rows_per_file, storage_options, data_storage_version.	{}

Returns:

Type	Description
LanceDataset | LazyFrame	The written lance.LanceDataset, or, when lazy is set, the
LanceDataset | LazyFrame	polars.LazyFrame that performs the write when it is collected.

Examples:

>>> query = lf.filter(pl.col("ok"))
>>> sink_lance(query, "out.lance", mode="overwrite")

Source code in src/polars_pylance/_sink.py

def sink_lance(
    lf: pl.LazyFrame | pl.DataFrame,
    target: str | Path | lance.LanceDataset,
    *,
    mode: WriteMode = "create",
    on: str | list[str] | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
    engine: EngineType = "streaming",
    lazy: bool = False,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> lance.LanceDataset | pl.LazyFrame:
    """Write a Polars DataFrame or stream a LazyFrame into a Lance dataset.

    A LazyFrame query is executed in batches and handed to Lance as it goes, so
    its result is never materialised in full. A DataFrame is accepted for
    convenience, but is already materialised before this function is called.

    Args:
        lf: The eager DataFrame or lazy query to write.
        target: Destination URI, path, or an existing `lance.LanceDataset`.
        mode: `"create"` (fail if it exists), `"append"`, `"overwrite"` (new version),
            or `"merge"` for an upsert, which requires `on`.
        on: Join key(s) for `mode="merge"`.
        chunk_size: Rows buffered per batch handed to Lance.
        engine: Polars engine. Leave at `"streaming"`; `"in-memory"` defeats the purpose
            by materialising the result first.
        lazy: Return a LazyFrame that performs the write when collected, instead of
            writing immediately, so the Lance write can sit inside a larger deferred
            plan. Collecting it yields a one-row summary of what was written.
        **lance_write_kwargs: Passed through to `lance.write_dataset`, e.g.
            `max_rows_per_file`, `storage_options`, `data_storage_version`.

    Returns:
        The written `lance.LanceDataset`, or, when `lazy` is set, the
        `polars.LazyFrame` that performs the write when it is collected.

    Examples:
        >>> query = lf.filter(pl.col("ok"))  # doctest: +SKIP
        >>> sink_lance(query, "out.lance", mode="overwrite")  # doctest: +SKIP
    """
    if isinstance(lf, pl.DataFrame):
        lf = lf.lazy()

    uri = _target_uri(target)

    if mode == "merge":
        if on is None:
            msg = "mode='merge' requires `on` (the join key column(s))"
            raise ValueError(msg)
        if lazy:
            msg = "mode='merge' does not support lazy=True"
            raise NotImplementedError(msg)
        if lance_write_kwargs:
            msg = (
                "mode='merge' does not accept write_dataset arguments: "
                f"{sorted(lance_write_kwargs)}"
            )
            raise TypeError(msg)
        dataset = (
            target
            if isinstance(target, lance.LanceDataset)
            else lance.dataset(uri, storage_options=None)
        )
        reader = _reader_from_lazyframe(lf, chunk_size=chunk_size, engine=engine)
        dataset.merge_insert(
            on
        ).when_matched_update_all().when_not_matched_insert_all().execute(reader)
        return dataset

    if on is not None:
        msg = f"`on` is only meaningful for mode='merge', not {mode!r}"
        raise ValueError(msg)

    if not lazy:
        reader = _reader_from_lazyframe(lf, chunk_size=chunk_size, engine=engine)
        return lance.write_dataset(reader, uri, mode=mode, **lance_write_kwargs)

    return _lazy_sink(
        lf,
        uri,
        mode=mode,
        chunk_size=chunk_size,
        engine=engine,
        lance_write_kwargs=lance_write_kwargs,
    )

write_lance_fragments

write_lance_fragments(lazyframes: Iterable[LazyFrame], target: str | Path, *, mode: Literal['create', 'overwrite', 'append'] = 'create', max_workers: int | None = None, chunk_size: int = DEFAULT_CHUNK_SIZE, engine: EngineType = 'streaming', arrow_schema: Schema | None = None, **lance_write_kwargs: Any) -> LanceDataset

Write several LazyFrames as Lance fragments in parallel, then commit once.

This is the distributed write shape: every shard streams into its own fragment files independently, and a single commit at the end makes them one dataset version. Pair with scan_lance_fragments to get the shards.

Parameters:

Name	Type	Description	Default
lazyframes	Iterable[LazyFrame]	One query per shard. All must produce the same schema.	required
target	str | Path	Destination URI or path.	required
mode	Literal['create', 'overwrite', 'append']	"create"/"overwrite" replace the dataset contents with the shards; "append" adds them to an existing dataset.	'create'
max_workers	int | None	Threads used to write shards. Defaults to one per shard.	None
chunk_size	int	Rows buffered per batch handed to Lance.	DEFAULT_CHUNK_SIZE
engine	EngineType	Polars engine. Leave at "streaming"; "in-memory" defeats the purpose by materialising each shard first.	'streaming'
arrow_schema	Schema | None	Schema to write. Inferred from the first shard when omitted.	None
**lance_write_kwargs	Any	Passed to lance.fragment.write_fragments.	{}

Examples:

>>> shards = scan_lance_fragments("in.lance")
>>> write_lance_fragments(
...     [s.filter(pl.col("ok")) for s in shards], "out.lance"
... )

Source code in src/polars_pylance/_sink.py

def write_lance_fragments(
    lazyframes: Iterable[pl.LazyFrame],
    target: str | Path,
    *,
    mode: Literal["create", "overwrite", "append"] = "create",
    max_workers: int | None = None,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
    engine: EngineType = "streaming",
    arrow_schema: pa.Schema | None = None,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> lance.LanceDataset:
    """Write several LazyFrames as Lance fragments in parallel, then commit once.

    This is the distributed write shape: every shard streams into its own fragment files
    independently, and a single commit at the end makes them one dataset version. Pair
    with [`scan_lance_fragments`][polars_pylance.scan_lance_fragments] to get the
    shards.

    Args:
        lazyframes: One query per shard. All must produce the same schema.
        target: Destination URI or path.
        mode: `"create"`/`"overwrite"` replace the dataset contents with the shards;
            `"append"` adds them to an existing dataset.
        max_workers: Threads used to write shards. Defaults to one per shard.
        chunk_size: Rows buffered per batch handed to Lance.
        engine: Polars engine. Leave at `"streaming"`; `"in-memory"` defeats the purpose
            by materialising each shard first.
        arrow_schema: Schema to write. Inferred from the first shard when omitted.
        **lance_write_kwargs: Passed to `lance.fragment.write_fragments`.

    Examples:
        >>> shards = scan_lance_fragments("in.lance")  # doctest: +SKIP
        >>> write_lance_fragments(
        ...     [s.filter(pl.col("ok")) for s in shards], "out.lance"
        ... )  # doctest: +SKIP
    """
    from concurrent.futures import ThreadPoolExecutor

    shards = list(lazyframes)
    if not shards:
        msg = "write_lance_fragments() needs at least one LazyFrame"
        raise ValueError(msg)

    uri = str(target)
    schema = arrow_schema or shards[0].collect_schema().to_arrow()

    def write_one(shard: pl.LazyFrame) -> list[FragmentMetadata]:
        return write_lance_shard(
            shard,
            uri,
            mode=mode,
            chunk_size=chunk_size,
            engine=engine,
            arrow_schema=schema,
            **lance_write_kwargs,
        )

    with ThreadPoolExecutor(max_workers=max_workers or len(shards)) as pool:
        written = list(pool.map(write_one, shards))

    fragments = [f for shard in written for f in shard]
    storage_options = lance_write_kwargs.get("storage_options")
    return commit_lance_fragments(
        uri, fragments, schema=schema, mode=mode, storage_options=storage_options
    )

write_lance_shard

write_lance_shard(shard: LazyFrame, target: str | Path, *, mode: Literal['create', 'overwrite', 'append'] = 'create', chunk_size: int = DEFAULT_CHUNK_SIZE, engine: EngineType = 'streaming', arrow_schema: Schema | None = None, **lance_write_kwargs: Any) -> list[FragmentMetadata]

Write one LazyFrame as Lance fragment files, without committing.

The single-shard primitive behind write_lance_fragments, factored out so distributed workers can call the same code the threaded path calls. It streams the query into fragment files and returns their metadata, publishing nothing: the caller commits with commit_lance_fragments once every shard is done.

Everything it takes is small and picklable -- a lazy query, a URI, a schema -- so a scheduler ships kilobytes per shard while the data itself moves only between the worker and Lance storage. The returned metadata is likewise small enough to send back to the coordinator.

Parameters:

Name	Type	Description	Default
shard	LazyFrame	The query to write. All shards committed together must produce the same schema.	required
target	str | Path	Destination URI or path. Workers writing to the same dataset must address it identically.	required
mode	Literal['create', 'overwrite', 'append']	"create"/"overwrite" write fragment files for a fresh schema; "append" reuses the existing dataset's field ids. Mirrors write_lance_fragments.	'create'
chunk_size	int	Rows buffered per batch handed to Lance.	DEFAULT_CHUNK_SIZE
engine	EngineType	Polars engine. Leave at "streaming"; "in-memory" defeats the purpose by materialising the shard first.	'streaming'
arrow_schema	Schema | None	Schema to write. Inferred from the shard when omitted; pass the coordinator's schema so every worker writes identically.	None
**lance_write_kwargs	Any	Passed to lance.fragment.write_fragments.	{}

Returns:

Type	Description
list[FragmentMetadata]	The lance.fragment.FragmentMetadata records for this shard's files.

Examples:

>>> shards = scan_lance_fragments("in.lance")
>>> write_lance_shard(
...     shards[0].filter(pl.col("ok")), "out.lance"
... )

Source code in src/polars_pylance/_sink.py

def write_lance_shard(
    shard: pl.LazyFrame,
    target: str | Path,
    *,
    mode: Literal["create", "overwrite", "append"] = "create",
    chunk_size: int = DEFAULT_CHUNK_SIZE,
    engine: EngineType = "streaming",
    arrow_schema: pa.Schema | None = None,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> list[FragmentMetadata]:
    """Write one LazyFrame as Lance fragment files, without committing.

    The single-shard primitive behind
    [`write_lance_fragments`][polars_pylance.write_lance_fragments], factored out
    so distributed workers can call the same code the threaded path calls. It
    streams the query into fragment files and returns their metadata, publishing
    nothing: the caller commits with
    [`commit_lance_fragments`][polars_pylance.commit_lance_fragments] once every
    shard is done.

    Everything it takes is small and picklable -- a lazy query, a URI, a schema
    -- so a scheduler ships kilobytes per shard while the data itself moves only
    between the worker and Lance storage. The returned metadata is likewise
    small enough to send back to the coordinator.

    Args:
        shard: The query to write. All shards committed together must produce
            the same schema.
        target: Destination URI or path. Workers writing to the same dataset
            must address it identically.
        mode: `"create"`/`"overwrite"` write fragment files for a fresh schema;
            `"append"` reuses the existing dataset's field ids. Mirrors
            [`write_lance_fragments`][polars_pylance.write_lance_fragments].
        chunk_size: Rows buffered per batch handed to Lance.
        engine: Polars engine. Leave at `"streaming"`; `"in-memory"` defeats the
            purpose by materialising the shard first.
        arrow_schema: Schema to write. Inferred from the shard when omitted;
            pass the coordinator's schema so every worker writes identically.
        **lance_write_kwargs: Passed to `lance.fragment.write_fragments`.

    Returns:
        The `lance.fragment.FragmentMetadata` records for this shard's files.

    Examples:
        >>> shards = scan_lance_fragments("in.lance")  # doctest: +SKIP
        >>> write_lance_shard(
        ...     shards[0].filter(pl.col("ok")), "out.lance"
        ... )  # doctest: +SKIP
    """
    uri = str(target)
    schema = arrow_schema or shard.collect_schema().to_arrow()
    fragment_mode = fragment_write_mode(mode)
    reader = _reader_from_lazyframe(shard, chunk_size=chunk_size, engine=engine)
    # `return_transaction=False` is the default; naming it picks the
    # overload that returns fragments rather than a transaction, which
    # `**lance_write_kwargs` would otherwise leave unresolved.
    return lance.fragment.write_fragments(
        reader,
        uri,
        schema=schema,
        mode=fragment_mode,
        return_transaction=False,
        **lance_write_kwargs,
    )