cloud

polars_pylance.cloud

Helpers for running Lance scans and writes on Polars Cloud.

Install the client with the cloud extra (pip install polars-pylance[cloud]), which brings polars-cloud>=0.11 tracking polars==1.44.2. See "The polars pin" below for why the extra exists rather than a hard dependency.

What works and what does not, as of polars-cloud 0.11:

Reading

A scan_lance plan serializes to ~2-6 kB and can be shipped with LazyFrame.remote(), but the remote workers must be able to import lance and to reach the dataset's storage. Install the dependency with ComputeContext(requirements=...); see requirements_txt.

The scan survives prepare_cloud_plan, on its own and under pl.concat of scan_lance_fragments shards. 0.9 added distributed unions of Python scans, so the sharded form is the sanctioned way to fan a read out across workers rather than a fallback.

Writing

Possible remotely since 0.10, via sink_lance_remote. Polars Cloud's native sink destinations are still Parquet, CSV, IPC and Iceberg, but sink_batches hands each result batch to a Python callable that is cloudpickled into the query plan and therefore runs on the workers, so the workers write Lance data files directly, and a single client-side commit publishes them. polars_pylance._remote documents the arrangement.

The Parquet-staging route remains as the conservative fallback: sink the remote query to Parquet on object storage and convert it with convert_parquet_to_lance. polars-cloud 0.10's DirectQuery.delete_result() makes cleaning up the intermediate a single call, in direct mode with anonymous storage configured for allow_delete.

The polars pin

polars-cloud pins polars with == (0.11 tracks polars==1.44.2, 0.10 required polars==1.43.2), and this package requires polars>=1.44.1. The cloud extra is an extra rather than a hard dependency so a plain pip install polars-pylance stays usable without a Cloud workspace; installing the extra resolves both to a 1.44 line that supports the IO-plugin hook scan_lance is built on.

The floor is there because 1.43.2 is the last release in which a sort().head() pushes an unevaluable dynamic_pred node into an IO plugin's predicate, which is exactly what scan_lance is. 1.44.0 fixed that but was yanked, so 1.44.1 is the first usable release.

StagedLanceSink dataclass

A remote Lance write in progress: a callback to ship, and a commit to run.

Returned by stage_lance_sink. Hand callback to sink_batches, wait for the query, then call commit.

Source code in src/polars_pylance/_remote.py

@dataclass
class StagedLanceSink:
    """A remote Lance write in progress: a callback to ship, and a commit to run.

    Returned by [`stage_lance_sink`][polars_pylance.cloud.stage_lance_sink]. Hand
    `callback` to `sink_batches`, wait for the query, then call
    [`commit`][polars_pylance.cloud.StagedLanceSink.commit].
    """

    uri: str
    staging_uri: str
    mode: RemoteWriteMode
    schema: pa.Schema
    callback: _FragmentWriter
    storage_options: dict[str, str] | None = None
    staging_filesystem: pafs.FileSystem | None = None
    staging_storage_options: dict[str, str] | None = None
    run_id: str = field(default_factory=lambda: uuid.uuid4().hex)

    def _fs(self) -> tuple[pafs.FileSystem, str]:
        return _resolve_filesystem(
            self.staging_uri,
            filesystem=self.staging_filesystem,
            storage_options=self.staging_storage_options,
        )

    def staged_fragments(self) -> list[Any]:
        """Every fragment the workers staged, in a deterministic order.

        Ordered by staging key so that two commits of the same staged output
        produce the same fragment layout.
        """
        fs, prefix = self._fs()
        selector = pafs.FileSelector(prefix, recursive=False, allow_not_found=True)
        entries: dict[str, list[Any]] = {}
        for info in fs.get_file_info(selector):
            if info.type != pafs.FileType.File or not info.path.endswith(".json"):
                continue
            with fs.open_input_stream(info.path) as stream:
                payload = json.loads(stream.readall())
            # Keyed by content, and the key is the object name, so a retried
            # batch has already overwritten itself. Re-keying here is belt and
            # braces against a caller-supplied `fragment_key` colliding.
            entries[payload["key"]] = payload["fragments"]

        return [
            lance.fragment.FragmentMetadata.from_json(json.dumps(fragment))
            for key in sorted(entries)
            for fragment in entries[key]
        ]

    def commit(self, *, cleanup: bool = True) -> lance.LanceDataset:
        """Make every staged fragment one dataset version, in a single commit.

        Args:
            cleanup: Remove the staging prefix once the commit lands. The *data* files a
                retried batch orphaned are not touched: they live inside the dataset and
                are unreferenced by any manifest, so reclaim them with
                `dataset.cleanup_old_versions(..., delete_unverified=True)`.

        Raises:
            ValueError: If nothing was staged. An empty commit would replace the dataset
                with nothing, which is never what a failed remote query meant.
        """
        fragments = self.staged_fragments()
        if not fragments:
            msg = (
                f"nothing staged under {self.staging_uri!r}: the remote query "
                "wrote no batches, or the workers could not reach the staging "
                "prefix (pass `staging_uri` / `staging_storage_options`)"
            )
            raise ValueError(msg)

        dataset = commit_lance_fragments(
            self.uri,
            fragments,
            schema=self.schema,
            mode=self.mode,
            storage_options=self.storage_options,
        )
        if cleanup:
            self.cleanup()
        return dataset

    def cleanup(self) -> None:
        """Delete the staging prefix. Safe to call when it was never created."""
        fs, prefix = self._fs()
        with contextlib.suppress(FileNotFoundError):
            fs.delete_dir(prefix)

staged_fragments

staged_fragments() -> list[Any]

Every fragment the workers staged, in a deterministic order.

Ordered by staging key so that two commits of the same staged output produce the same fragment layout.

Source code in src/polars_pylance/_remote.py

def staged_fragments(self) -> list[Any]:
    """Every fragment the workers staged, in a deterministic order.

    Ordered by staging key so that two commits of the same staged output
    produce the same fragment layout.
    """
    fs, prefix = self._fs()
    selector = pafs.FileSelector(prefix, recursive=False, allow_not_found=True)
    entries: dict[str, list[Any]] = {}
    for info in fs.get_file_info(selector):
        if info.type != pafs.FileType.File or not info.path.endswith(".json"):
            continue
        with fs.open_input_stream(info.path) as stream:
            payload = json.loads(stream.readall())
        # Keyed by content, and the key is the object name, so a retried
        # batch has already overwritten itself. Re-keying here is belt and
        # braces against a caller-supplied `fragment_key` colliding.
        entries[payload["key"]] = payload["fragments"]

    return [
        lance.fragment.FragmentMetadata.from_json(json.dumps(fragment))
        for key in sorted(entries)
        for fragment in entries[key]
    ]

commit

commit(*, cleanup: bool = True) -> LanceDataset

Make every staged fragment one dataset version, in a single commit.

Parameters:

Name	Type	Description	Default
cleanup	bool	Remove the staging prefix once the commit lands. The data files a retried batch orphaned are not touched: they live inside the dataset and are unreferenced by any manifest, so reclaim them with dataset.cleanup_old_versions(..., delete_unverified=True).	True

Raises:

Type	Description
ValueError	If nothing was staged. An empty commit would replace the dataset with nothing, which is never what a failed remote query meant.

Source code in src/polars_pylance/_remote.py

def commit(self, *, cleanup: bool = True) -> lance.LanceDataset:
    """Make every staged fragment one dataset version, in a single commit.

    Args:
        cleanup: Remove the staging prefix once the commit lands. The *data* files a
            retried batch orphaned are not touched: they live inside the dataset and
            are unreferenced by any manifest, so reclaim them with
            `dataset.cleanup_old_versions(..., delete_unverified=True)`.

    Raises:
        ValueError: If nothing was staged. An empty commit would replace the dataset
            with nothing, which is never what a failed remote query meant.
    """
    fragments = self.staged_fragments()
    if not fragments:
        msg = (
            f"nothing staged under {self.staging_uri!r}: the remote query "
            "wrote no batches, or the workers could not reach the staging "
            "prefix (pass `staging_uri` / `staging_storage_options`)"
        )
        raise ValueError(msg)

    dataset = commit_lance_fragments(
        self.uri,
        fragments,
        schema=self.schema,
        mode=self.mode,
        storage_options=self.storage_options,
    )
    if cleanup:
        self.cleanup()
    return dataset

cleanup

cleanup() -> None

Delete the staging prefix. Safe to call when it was never created.

Source code in src/polars_pylance/_remote.py

def cleanup(self) -> None:
    """Delete the staging prefix. Safe to call when it was never created."""
    fs, prefix = self._fs()
    with contextlib.suppress(FileNotFoundError):
        fs.delete_dir(prefix)

sink_lance_remote

sink_lance_remote(remote: LazyFrameRemote | ExecuteRemote, target: str | Path | LanceDataset, *, schema: Schema | Schema | None = None, mode: RemoteWriteMode = 'create', chunk_size: int | None = None, maintain_order: bool = False, storage_options: dict[str, str] | None = None, staging_uri: str | None = None, staging_filesystem: FileSystem | None = None, staging_storage_options: dict[str, str] | None = None, fragment_key: Callable[[DataFrame], str] | None = None, cleanup: bool = True, **lance_write_kwargs: Any) -> LanceDataset

Run a Polars Cloud query and write its output to Lance from the workers.

The staging parameters are documented on stage_lance_sink, which receives them unchanged.

Submits remote with a worker-side fragment writer, waits for it, and commits every staged fragment as one dataset version.

Parameters:

Name	Type	Description	Default
remote	LazyFrameRemote | ExecuteRemote	A polars_cloud.LazyFrameRemote (the result of lf.remote(ctx)) or the ExecuteRemote that .distributed() / .single_node() return.	required
target	str | Path | LanceDataset	Destination URI, path, or an existing lance.LanceDataset.	required
schema	Schema | Schema | None	The query's output schema. Resolved from the LazyFrame behind remote when omitted, which requires the client to be able to reach the sources.	None
chunk_size	int | None	Rows buffered before the callback runs; the remote counterpart of max_rows_per_file. Left to Polars Cloud when omitted.	None
maintain_order	bool	Call the writer serially rather than in parallel across workers. Costs the parallelism that makes this a distributed write; the fragment order at commit is deterministic either way.	False
cleanup	bool	Remove the staging prefix after a successful commit. On failure it is always left in place, so a re-run can be diagnosed or the fragments committed by hand.	True

Returns:

Type	Description
LanceDataset	lance.LanceDataset: The committed dataset.

Examples:

>>> ctx = pc.ComputeContext(
...     cpus=8, memory=32, requirements=requirements_txt().encode()
... )
>>> lf = scan_lance("s3://bucket/in.lance").filter(
...     pl.col("score") > 0.9
... )
>>> sink_lance_remote(
...     lf.remote(ctx).distributed(), "s3://bucket/out.lance", mode="overwrite"
... )

Source code in src/polars_pylance/_remote.py

def sink_lance_remote(  # noqa: D417 - the staging parameters are documented once, on `stage_lance_sink`
    remote: LazyFrameRemote | ExecuteRemote,
    target: str | Path | lance.LanceDataset,
    *,
    schema: pa.Schema | pl.Schema | None = None,
    mode: RemoteWriteMode = "create",
    chunk_size: int | None = None,
    maintain_order: bool = False,
    storage_options: dict[str, str] | None = None,
    staging_uri: str | None = None,
    staging_filesystem: pafs.FileSystem | None = None,
    staging_storage_options: dict[str, str] | None = None,
    fragment_key: Callable[[pl.DataFrame], str] | None = None,
    cleanup: bool = True,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> lance.LanceDataset:
    """Run a Polars Cloud query and write its output to Lance from the workers.

    The staging parameters are documented on
    [`stage_lance_sink`][polars_pylance.cloud.stage_lance_sink], which receives them
    unchanged.

    Submits `remote` with a worker-side fragment writer, waits for it, and
    commits every staged fragment as one dataset version.

    Args:
        remote: A `polars_cloud.LazyFrameRemote` (the result of `lf.remote(ctx)`) or the
            `ExecuteRemote` that `.distributed()` / `.single_node()` return.
        target: Destination URI, path, or an existing `lance.LanceDataset`.
        schema: The query's output schema. Resolved from the LazyFrame behind `remote`
            when omitted, which requires the client to be able to reach the sources.
        chunk_size: Rows buffered before the callback runs; the remote counterpart of
            `max_rows_per_file`. Left to Polars Cloud when omitted.
        maintain_order: Call the writer serially rather than in parallel across workers.
            Costs the parallelism that makes this a distributed write; the fragment
            order at commit is deterministic either way.
        cleanup: Remove the staging prefix after a successful commit. On failure it is
            always left in place, so a re-run can be diagnosed or the fragments
            committed by hand.

    Returns:
        lance.LanceDataset: The committed dataset.

    Examples:
        >>> ctx = pc.ComputeContext(
        ...     cpus=8, memory=32, requirements=requirements_txt().encode()
        ... )  # doctest: +SKIP
        >>> lf = scan_lance("s3://bucket/in.lance").filter(
        ...     pl.col("score") > 0.9
        ... )  # doctest: +SKIP
        >>> sink_lance_remote(
        ...     lf.remote(ctx).distributed(), "s3://bucket/out.lance", mode="overwrite"
        ... )  # doctest: +SKIP
    """
    if schema is None:
        schema = remote.lf.collect_schema()

    staged = stage_lance_sink(
        target,
        schema,
        mode=mode,
        storage_options=storage_options,
        staging_uri=staging_uri,
        staging_filesystem=staging_filesystem,
        staging_storage_options=staging_storage_options,
        fragment_key=fragment_key,
        **lance_write_kwargs,
    )

    query = remote.sink_batches(
        staged.callback, chunk_size=chunk_size, maintain_order=maintain_order
    )
    query.await_result()

    return staged.commit(cleanup=cleanup)

stage_lance_sink

stage_lance_sink(target: str | Path | LanceDataset, schema: Schema | Schema | LazyFrame, *, mode: RemoteWriteMode = 'create', storage_options: dict[str, str] | None = None, staging_uri: str | None = None, staging_filesystem: FileSystem | None = None, staging_storage_options: dict[str, str] | None = None, fragment_key: Callable[[DataFrame], str] | None = None, **lance_write_kwargs: Any) -> StagedLanceSink

Build a worker-side Lance writer for sink_batches.

Use this when you want to drive the remote query yourself: to pick a planner, set maintain_order, or inspect the query handle. sink_lance_remote is the same thing with the query submitted and awaited for you.

Parameters:

Name	Type	Description	Default
target	str | Path | LanceDataset	Destination URI, path, or an existing lance.LanceDataset.	required
schema	Schema | Schema | LazyFrame	The query's output schema. A LazyFrame is accepted and resolved with collect_schema(), which needs the client to be able to reach the sources, as it already must be to build a scan_lance plan.	required
mode	RemoteWriteMode	"create" (fail if the dataset exists), "overwrite" (replace its contents with a new version), or "append".	'create'
storage_options	dict[str, str] | None	Passed to Lance for the data files, and (for s3:// targets) translated into a PyArrow filesystem for the staging prefix.	None
staging_uri	str | None	Where fragment metadata is staged. Defaults to the dataset URI plus .pll-staging, with a per-run subdirectory so concurrent writes to one dataset do not read each other's fragments.	None
staging_filesystem	FileSystem | None	An explicit pyarrow.fs.FileSystem for the staging prefix, for stores whose credentials do not translate. Client-side only: the callback resolves the staging filesystem itself on the worker, from staging_storage_options or the worker's ambient credentials.	None
staging_storage_options	dict[str, str] | None	Staging credentials, when they differ from storage_options.	None
fragment_key	Callable[[DataFrame], str] | None	(batch) -> str, replacing the default content digest. Must be deterministic: two invocations for the same batch must agree, and two distinct batches must not. Worth supplying when the query carries a natural key, such as a partition column.	None
**lance_write_kwargs	Any	Passed to lance.fragment.write_fragments, e.g. max_rows_per_file, data_storage_version.	{}

Examples:

>>> staged = stage_lance_sink(
...     "s3://bucket/out.lance", lf, mode="overwrite"
... )
>>> query = (
...     lf.remote(ctx).distributed().sink_batches(staged.callback)
... )
>>> query.await_result()
>>> staged.commit()

Source code in src/polars_pylance/_remote.py

def stage_lance_sink(
    target: str | Path | lance.LanceDataset,
    schema: pa.Schema | pl.Schema | pl.LazyFrame,
    *,
    mode: RemoteWriteMode = "create",
    storage_options: dict[str, str] | None = None,
    staging_uri: str | None = None,
    staging_filesystem: pafs.FileSystem | None = None,
    staging_storage_options: dict[str, str] | None = None,
    fragment_key: Callable[[pl.DataFrame], str] | None = None,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> StagedLanceSink:
    """Build a worker-side Lance writer for `sink_batches`.

    Use this when you want to drive the remote query yourself: to pick a planner, set
    `maintain_order`, or inspect the query handle.
    [`sink_lance_remote`][polars_pylance.cloud.sink_lance_remote] is the same thing with
    the query submitted and awaited for you.

    Args:
        target: Destination URI, path, or an existing `lance.LanceDataset`.
        schema: The query's output schema. A LazyFrame is accepted and resolved with
            `collect_schema()`, which needs the *client* to be able to reach the
            sources, as it already must be to build a `scan_lance` plan.
        mode: `"create"` (fail if the dataset exists), `"overwrite"` (replace its
            contents with a new version), or `"append"`.
        storage_options: Passed to Lance for the data files, and (for `s3://` targets)
            translated into a PyArrow filesystem for the staging prefix.
        staging_uri: Where fragment metadata is staged. Defaults to the dataset URI plus
            `.pll-staging`, with a per-run subdirectory so concurrent writes to one
            dataset do not read each other's fragments.
        staging_filesystem: An explicit [`pyarrow.fs.FileSystem`][pyarrow.fs.FileSystem]
            for the staging prefix, for stores whose credentials do not translate.
            Client-side only: the callback resolves the staging filesystem itself on the
            worker, from `staging_storage_options` or the worker's ambient credentials.
        staging_storage_options: Staging credentials, when they differ from
            `storage_options`.
        fragment_key: `(batch) -> str`, replacing the default content digest. Must be
            deterministic: two invocations for the same batch must agree, and two
            distinct batches must not. Worth supplying when the query carries a natural
            key, such as a partition column.
        **lance_write_kwargs: Passed to `lance.fragment.write_fragments`, e.g.
            `max_rows_per_file`, `data_storage_version`.

    Examples:
        >>> staged = stage_lance_sink(
        ...     "s3://bucket/out.lance", lf, mode="overwrite"
        ... )  # doctest: +SKIP
        >>> query = (
        ...     lf.remote(ctx).distributed().sink_batches(staged.callback)
        ... )  # doctest: +SKIP
        >>> query.await_result()  # doctest: +SKIP
        >>> staged.commit()  # doctest: +SKIP
    """
    uri = target.uri if isinstance(target, lance.LanceDataset) else str(target)
    arrow_schema = _as_arrow_schema(schema)
    run_id = uuid.uuid4().hex
    base = staging_uri if staging_uri is not None else uri.rstrip("/") + STAGING_SUFFIX
    run_staging = f"{base.rstrip('/')}/{run_id}"

    if staging_storage_options is None and storage_options is not None:
        staging_storage_options = storage_options

    # Checked again at commit time, which is authoritative. Doing it here too
    # means `mode="create"` over an existing dataset costs nothing rather than a
    # whole cluster run that is discarded.
    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)

    fragment_mode = fragment_write_mode(mode)

    callback = _FragmentWriter(
        uri=uri,
        schema_ipc=arrow_schema.serialize().to_pybytes(),
        staging_uri=run_staging,
        fragment_mode=fragment_mode,
        storage_options=storage_options,
        staging_storage_options=staging_storage_options,
        write_kwargs=dict(lance_write_kwargs),
        fragment_key=fragment_key,
    )

    return StagedLanceSink(
        uri=uri,
        staging_uri=run_staging,
        mode=mode,
        schema=arrow_schema,
        callback=callback,
        storage_options=storage_options,
        staging_filesystem=staging_filesystem,
        staging_storage_options=staging_storage_options,
        run_id=run_id,
    )

requirements_txt

requirements_txt(extra: list[str] | None = None) -> str

Render a requirements file pinning the versions a cloud worker needs.

Polars Cloud rejects a compute context whose polars version differs from the client's, so both pins are exact. polars-pylance itself is on the list because a sink_lance_remote callback is pickled by reference: the worker imports it rather than receiving its code.

Parameters:

Name	Type	Description	Default
extra	list[str] | None	Further requirement lines to append, for whatever else the query needs on the worker.	None

Returns:

Name	Type	Description
str	str	The file contents, newline-terminated.

Examples:

>>> import polars_cloud as pc
>>> ctx = pc.ComputeContext(
...     cpus=8, memory=32, requirements=requirements_txt().encode()
... )

Source code in src/polars_pylance/cloud.py

def requirements_txt(extra: list[str] | None = None) -> str:
    """Render a requirements file pinning the versions a cloud worker needs.

    Polars Cloud rejects a compute context whose polars version differs from the
    client's, so both pins are exact. `polars-pylance` itself is on the list because a
    [`sink_lance_remote`][polars_pylance.cloud.sink_lance_remote] callback is pickled by
    reference: the worker imports it rather than receiving its code.

    Args:
        extra: Further requirement lines to append, for whatever else the query needs on
            the worker.

    Returns:
        str: The file contents, newline-terminated.

    Examples:
        >>> import polars_cloud as pc  # doctest: +SKIP
        >>> ctx = pc.ComputeContext(
        ...     cpus=8, memory=32, requirements=requirements_txt().encode()
        ... )  # doctest: +SKIP
    """
    lines = [
        f"polars=={pl.__version__}",
        f"pylance=={lance.__version__}",
        "polars-pylance",
    ]
    lines.extend(extra or [])
    return "\n".join(lines) + "\n"

convert_parquet_to_lance

convert_parquet_to_lance(parquet_source: str | Path | list[str], target: str | Path, *, mode: WriteMode = 'create', chunk_size: int = 25000, storage_options: dict[str, str] | None = None, **lance_write_kwargs: Any) -> LanceDataset

Stream Parquet output from a remote query into a Lance dataset.

The documented way to land Polars Cloud results in Lance: the remote query sinks Parquet to object storage, then this converts it without materialising the data.

Parameters:

Name	Type	Description	Default
parquet_source	str | Path | list[str]	The staged Parquet: a path, a URI, a glob, or a list of them.	required
target	str | Path	Destination Lance URI or path.	required
mode	WriteMode	"create" (fail if it exists), "append", "overwrite" (new version), or "merge" for an upsert, whose join key goes through lance_write_kwargs as on.	'create'
chunk_size	int	Rows buffered per batch handed to Lance.	25000
storage_options	dict[str, str] | None	Object-store credentials and settings for reading the Parquet. The Lance write takes its own; pass those in lance_write_kwargs.	None
**lance_write_kwargs	Any	Passed through to sink_lance, e.g. max_rows_per_file.	{}

Returns:

Type	Description
LanceDataset	lance.LanceDataset: The written dataset.

Examples:

>>> query.remote(ctx).distributed().sink_parquet(staging)
>>> convert_parquet_to_lance(staging, "s3://bucket/out.lance")

Source code in src/polars_pylance/cloud.py

def convert_parquet_to_lance(
    parquet_source: str | Path | list[str],
    target: str | Path,
    *,
    mode: WriteMode = "create",
    chunk_size: int = 25_000,
    storage_options: dict[str, str] | None = None,
    **lance_write_kwargs: Any,  # noqa: ANN401 - passed through to Lance as given
) -> lance.LanceDataset:
    """Stream Parquet output from a remote query into a Lance dataset.

    The documented way to land Polars Cloud results in Lance: the remote query
    sinks Parquet to object storage, then this converts it without materialising
    the data.

    Args:
        parquet_source: The staged Parquet: a path, a URI, a glob, or a list of them.
        target: Destination Lance URI or path.
        mode: `"create"` (fail if it exists), `"append"`, `"overwrite"` (new version),
            or `"merge"` for an upsert, whose join key goes through `lance_write_kwargs`
            as `on`.
        chunk_size: Rows buffered per batch handed to Lance.
        storage_options: Object-store credentials and settings for reading the Parquet.
            The Lance write takes its own; pass those in `lance_write_kwargs`.
        **lance_write_kwargs: Passed through to
            [`sink_lance`][polars_pylance.sink_lance], e.g. `max_rows_per_file`.

    Returns:
        lance.LanceDataset: The written dataset.

    Examples:
        >>> query.remote(ctx).distributed().sink_parquet(staging)  # doctest: +SKIP
        >>> convert_parquet_to_lance(staging, "s3://bucket/out.lance")  # doctest: +SKIP
    """
    from ._sink import sink_lance

    lf = pl.scan_parquet(parquet_source, storage_options=storage_options)
    dataset = sink_lance(
        lf,
        target,
        mode=mode,
        chunk_size=chunk_size,
        **lance_write_kwargs,
    )
    assert isinstance(dataset, lance.LanceDataset)
    return dataset