pola-rs/polars · error · NotImplementedError

`sink_batches` is not supported by {type(self).__name__}

Error message

`sink_batches` is not supported by {type(self).__name__}

What it means

`Engine.sink_batches` (engine.py:341) is the optional `Engine` hook behind `LazyFrame.sink_batches`, which streams result batches to a Python callback `function(DataFrame) -> bool | None`. Only the local `_LocalEngine` family implements it; `RemoteEngine` does not, and neither do minimal custom engines. The stub raises `NotImplementedError` naming the engine class.

Source

Thrown at py-polars/src/polars/lazyframe/engine.py:341

        optimizations: QueryOptFlags,
    ) -> LazyFrame | None:
        """See :meth:`polars.LazyFrame.sink_ndjson`."""
        msg = f"`sink_ndjson` is not supported by {type(self).__name__}"
        raise NotImplementedError(msg)

    def sink_batches(
        self,
        lf: LazyFrame,
        function: Callable[[DataFrame], bool | None],
        *,
        chunk_size: int | None,
        maintain_order: bool,
        lazy: bool,
        optimizations: QueryOptFlags,
    ) -> LazyFrame | None:
        """See :meth:`polars.LazyFrame.sink_batches`."""
        msg = f"`sink_batches` is not supported by {type(self).__name__}"
        raise NotImplementedError(msg)


class _LocalEngine(Engine):
    """Base for in-process engines backed by `PyLazyFrame`."""

    _name: ClassVar[str]

    @property
    def name(self) -> str:
        """Name of the engine."""
        return self._name

    def execute(self, lf: LazyFrame, *, optimizations: QueryOptFlags) -> QueryResult:
        df = self.collect(lf, optimizations=optimizations)
        return SingleNodeQueryResult(df)  # type: ignore[arg-type]

    def _post_opt_callback(
        self,

View on GitHub (pinned to df599052da)

Solutions

1. Use a local engine: `lf.sink_batches(fn, engine='streaming')` (default)
2. For remote data movement, sink to a cloud URI with `sink_parquet`/`sink_ipc`/`sink_csv` on RemoteEngine and process the files separately
3. If you own the engine, implement `sink_batches` with the stub's signature (function, chunk_size, maintain_order, lazy, optimizations)
4. Collect first as a stopgap: iterate `lf.collect().iter_slices(n)`

Example fix

# before
lf.sink_batches(consume, engine=pl.RemoteEngine())  # NotImplementedError

# after
lf.sink_batches(consume, engine='streaming')

from polars.lazyframe.engine import Engine

def engine_can_sink_batches(engine: pl.Engine) -> bool:
    return type(engine).sink_batches is not Engine.sink_batches

engine = engine if engine_can_sink_batches(engine) else pl.StreamingEngine()
lf.sink_batches(consume, engine=engine)

try:
    lf.sink_batches(fn, engine=engine)
except NotImplementedError as e:
    if 'sink_batches' in str(e):
        for chunk in lf.collect(engine=engine).iter_slices(100_000):
            fn(chunk)
    else:
        raise

When it happens

Trigger: `lf.sink_batches(fn, engine=pl.RemoteEngine())` or `lf.sink_batches(fn, engine=my_engine)` with a custom `Engine` subclass that implements only `collect`/`execute`. Because the callback runs in-process, no remote implementation exists.

Common situations: Feeding query output chunk-by-chunk into another in-process system (vector DBs, HTTP uploaders) and then pointing the query at Polars Cloud; custom backends where batch callbacks were never wired; shared utility code that assumes every engine accepts `sink_batches`.

Related errors

• `sink_ndjson` is not supported by {type(self).__name__}
• `sink_parquet` is not supported by {type(self).__name__}
• `sink_ipc` is not supported by {type(self).__name__}
• `sink_csv` is not supported by {type(self).__name__}
• `collect_async` is not supported by {type(self).__name__}

AI-assisted analysis of pola-rs/polars@df599052da (2026-08-16). Data as JSON: /api/errors/31813e95cad02b62. Report an issue: GitHub.