pola-rs/polars · error · NotImplementedError

`collect_all_async` is not supported by

Error message

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

What it means

`collect_all_async` executes several LazyFrames asynchronously and is only available on engines implementing it. The base `Engine` stub raises NotImplementedError with this message, so invoking it on a non-async-capable engine fails.

Solutions

Call `collect_all_async` without overriding the engine (use the async-capable default).
Fall back to synchronous `pl.collect_all(lfs)` wrapped in a thread/executor for non-blocking behavior.
Implement `collect_all_async` in your custom Engine returning `AsyncResult[list[DataFrame]]`.
Check engine capabilities before mixing async APIs with custom engines.

Example fix

// before
await pl.collect_all_async(lfs, engine="custom")
// after
dfs = pl.collect_all(lfs)

Defensive patterns

Strategy: try-catch

Validate before calling

if type(engine_obj).collect_all_async is Engine.collect_all_async:
    use_sync_collect_all = True

Type guard

def supports_async_all(engine: object) -> bool:
    return type(engine).collect_all_async is not Engine.collect_all_async

Try / catch

try:
    res = pl.collect_all_async(lfs, engine=eng)
except NotImplementedError:
    dfs = pl.collect_all(lfs)

Prevention

Use the async-capable default engine for collect_all_async
Implement the method when subclassing Engine
Guard async batch APIs with fallbacks

When it happens

Trigger: Calling `pl.collect_all_async(lfs, engine=<engine>)` with an engine that does not override `collect_all_async` (e.g. custom engine or base implementation reached).

Common situations: Async pipelines collecting many frames concurrently with a custom engine; examples written for the async in-process engine run against a different engine selection.

Related errors

AI-assisted analysis of pola-rs/polars@fe841f959e (2026-09-18). Data as JSON: /api/errors/3250a3395cc823fa. Report an issue: GitHub.

Appendix: source

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

        raise NotImplementedError(msg)

    def collect_all(
        self, lfs: Iterable[LazyFrame], *, optimizations: QueryOptFlags
    ) -> list[DataFrame]:
        """Execute several queries, potentially in parallel."""
        msg = f"`collect_all` is not supported by {type(self).__name__}"
        raise NotImplementedError(msg)

    def collect_all_async(
        self,
        lfs: Iterable[LazyFrame],
        *,
        optimizations: QueryOptFlags,
        gevent: bool = False,
    ) -> AsyncResult[list[DataFrame]]:
        """Execute several queries asynchronously."""
        msg = f"`collect_all_async` is not supported by {type(self).__name__}"
        raise NotImplementedError(msg)

    def sink_parquet(
        self,
        lf: LazyFrame,
        path: str | Path | IO[bytes] | PartitionBy,
        *,
        compression: ParquetCompression,
        compression_level: int | None,
        statistics: bool | str | dict[str, bool],
        row_group_size: int | None,
        data_page_size: int | None,
        maintain_order: bool,
        storage_options: StorageOptionsDict | None,
        credential_provider: CredentialProviderFunction | Literal["auto"] | None,
        sync_on_close: SyncOnCloseMethod | None,
        metadata: ParquetMetadata | None,
        arrow_schema: ArrowSchemaExportable | None,
        mkdir: bool,

View on GitHub (pinned to fe841f959e)