API Reference#

For the most part, the public API of cudf-polars is the Polars API itself. This page documents the additional classes and functions that cudf-polars exposes for the streaming multi-GPU engines.

Streaming engines#

class cudf_polars.engine.ray.RayEngine( *, rapidsmpf_options: Options | None = None, executor_options: dict[str, Any] | None = None, engine_options: dict[str, Any] | None = None, ray_init_options: dict[str, Any] | None = None, num_ranks: int | None = None, )[source]#

Bases: StreamingEngine

Multi-GPU Polars engine for Ray cluster execution.

Creates a RapidsMPF Ray cluster and returns an engine that can be passed to LazyFrame.collect(engine=engine).

Prefer from_options() for typical use. Pass a StreamingOptions instance for a unified, typed interface. The __init__ parameters (rapidsmpf_options, executor_options, engine_options) are intended for advanced use when fine-grained control is needed.

Prefer the context-manager form in scripts: it guarantees that actors and Ray are shut down even if an exception is raised. In interactive environments such as Jupyter notebooks, the direct form lets the cluster persist across multiple cells without tearing it down after every query.

If Ray is not already initialized, ray.init() is called here and ray.shutdown() is called by shutdown(). If Ray is already initialized, cluster lifetime remains managed by the caller.

Parameters:

rapidsmpf_options

RapidsMPF-specific options. Defaults to the reading RAPIDSMPF_* environment variables.

executor_options

Executor-specific options (e.g. max_rows_per_partition).

engine_options

Engine-specific keyword arguments (e.g. raise_on_fail, parquet_options).

ray_init_options

Keyword arguments forwarded to ray.init() when Ray is not already initialized.

num_ranks

Number of ranks (Ray actors) to create. When None (the default), one rank is created per available GPU using Ray’s GPU scheduling, which provides placement guarantees and topology-aware hardware binding. When set, bypasses Ray’s GPU resource accounting so that actors do not contend for GPU resource slots. This allows multiple RayEngine instances to share a single Ray cluster and enables oversubscribed execution on limited GPU hardware. Hardware binding is disabled implicitly but the caller must pass engine_options={"allow_gpu_sharing": True} explicitly to acknowledge the multi-tenant GPU semantics.

Note, oversubscription does not increase throughput. When multiple ranks share a GPU, they compete for the same compute and memory resources, which may increase memory pressure and reduce overall performance. This option is primarily useful for testing multi-rank code paths on machines with fewer GPUs than ranks, and for downstream projects that need to validate distributed logic in resource-constrained CI environments.

Attributes

nranks

Number of ranks (for example GPUs or workers) in the cluster.

Methods

`from_options`(options, *[, ray_init_options])	Create a `RayEngine` from a `StreamingOptions` object.
`gather_cluster_info`()	Collect diagnostic information from every rank.
`gather_statistics`(*[, clear])	Collect statistics from every rank via Ray.
`global_statistics`(*[, clear])	Collect statistics from every rank and merge them into a single global statistics.
`shutdown`()	Shut down all rank actors and release resources.

Raises:

RuntimeError: If called from within an rrun cluster.
RuntimeError: If not all GPUs in the Ray cluster are free at startup (only when num_ranks is None).
RuntimeError: If no GPUs are available in the Ray cluster (only when num_ranks is None).
TypeError: If executor_options or engine_options contains a reserved key.
ValueError: If num_ranks is set but engine_options["allow_gpu_sharing"] == False
ValueError: If num_ranks is set to a value less than 1.

Examples

Context-manager style:

>>> with RayEngine() as engine:
...     result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)

Jupyter / manual style:

>>> engine = RayEngine()
>>> result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)
>>> engine.shutdown()

classmethod from_options( options: StreamingOptions, *, ray_init_options: dict[str, object] | None = None, ) → RayEngine[source]#

Create a RayEngine from a StreamingOptions object.

This is the recommended way to construct a RayEngine for typical use. All RapidsMPF, executor, and engine options are read from options; unset fields fall back to environment variables and then to built-in defaults.

Parameters:

options: Unified streaming configuration.
ray_init_options: Keyword arguments forwarded to ray.init() when Ray is not already initialized. These are Ray infrastructure settings and are kept separate from streaming behavior options.

Returns:

A new RayEngine instance.

Examples

>>> from cudf_polars.engine.options import StreamingOptions
>>> opts = StreamingOptions(num_streaming_threads=4, fallback_mode="silent")
>>> with RayEngine.from_options(opts) as engine:
...     result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)

gather_cluster_info() → list[ClusterInfo][source]#

Collect diagnostic information from every rank.

Returns:

List of ClusterInfo, one per rank.

gather_statistics( *, clear: bool = False, ) → list[Statistics][source]#

Collect statistics from every rank via Ray.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

List of Statistics, one per rank,
ordered by rank index.

global_statistics( *, clear: bool = False, ) → Statistics[source]#

Collect statistics from every rank and merge them into a single global statistics.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

A merged Statistics: per-stat counts
and values are summed, maxima are reduced with max. Formatters
are taken from rank 0.

property nranks: int[source]#

Number of ranks (for example GPUs or workers) in the cluster.

Local execution without a cluster returns 1.

Returns:

Number of ranks.

shutdown() → None[source]#

Shut down all rank actors and release resources.

If Ray was initialized by this engine, also calls ray.shutdown(). Safe to call more than once.

Raises:

ExceptionGroup: If one or more actors raise an unexpected exception during shutdown.

class cudf_polars.engine.dask.DaskEngine( *, dask_client: Client | None = None, rapidsmpf_options: Options | None = None, executor_options: dict[str, Any] | None = None, engine_options: dict[str, Any] | None = None, )[source]#

Bases: StreamingEngine

Multi-GPU Polars engine for Dask distributed execution backed by RapidsMPF.

Bootstraps a RapidsMPF UCXX cluster on top of a Dask distributed cluster and returns an engine that can be passed to LazyFrame.collect(engine=engine).

If dask_client is provided, it is used directly and its lifetime is managed by the caller. If dask_client is None, a distributed.LocalCluster is created automatically (one worker per visible GPU) and torn down by shutdown().

Prefer the context-manager form in scripts: it guarantees that workers are torn down even if an exception is raised. In interactive environments such as Jupyter notebooks, the direct form lets the cluster persist across multiple cells without tearing it down after every query.

Parameters:

dask_client: An existing Client to use. If None, a LocalCluster (one worker per visible GPU) and a new client are created and owned by this engine.
rapidsmpf_options: RapidsMPF options forwarded to every worker. If None, defaults to Options(get_environment_variables()).
executor_options: Executor-specific options (e.g. max_rows_per_partition).
engine_options: Engine-specific keyword arguments (e.g. raise_on_fail, parquet_options).

Attributes

nranks

Number of ranks (for example GPUs or workers) in the cluster.

Methods

`from_options`(options, *[, dask_client])	Create a `DaskEngine` from a `StreamingOptions` object.
`gather_cluster_info`()	Collect diagnostic information from every rank.
`gather_statistics`(*[, clear])	Collect statistics from every rank via `client.run`.
`global_statistics`(*[, clear])	Collect statistics from every rank and merge them into a single global statistics.
`shutdown`()	Shut down all Dask workers' GPU resources.

Raises:

RuntimeError: If called from within an rrun cluster.
TypeError: If executor_options or engine_options contains a reserved key.

Notes

When using a pre-configured cluster that already performs its own hardware binding (e.g. dask_cuda.LocalCUDACluster, which pins CPU affinity and sets CUDA_VISIBLE_DEVICES per worker), disable some or all of the built-in binding to avoid conflicts:

>>> from cudf_polars.engine.hardware_binding import HardwareBindingPolicy
>>> with DaskEngine(
...     dask_client=dc,
...     engine_options={
...         "hardware_binding": HardwareBindingPolicy(enabled=False),
...     },
... ) as engine:
...     ...

For manually launched Dask clusters, use the nanny preload to assign one GPU per worker before the worker process spawns:

dask worker SCHEDULER:8786 --nworkers N --nthreads 1             --preload-nanny cudf_polars.engine.dask

Then connect from the client:

>>> from distributed import Client
>>> with Client("SCHEDULER:8786") as dc:
...     with DaskEngine(dask_client=dc) as engine:
...         result = lf.collect(engine=engine)

Examples

Context-manager style:

>>> with DaskEngine() as engine:
...     result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)

Bring-your-own client:

>>> from distributed import Client
>>> with Client("scheduler-address:8786") as dc:
...     with DaskEngine(dask_client=dc) as engine:
...         result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)

Jupyter / manual style:

>>> engine = DaskEngine()
>>> result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)
>>> engine.shutdown()

classmethod from_options( options: StreamingOptions, *, dask_client: distributed.Client | None = None, ) → DaskEngine[source]#

Create a DaskEngine from a StreamingOptions object.

This is the recommended way to construct a DaskEngine for typical use. All RapidsMPF, executor, and engine options are read from options; unset fields fall back to environment variables and then to built-in defaults.

Parameters:

options: Unified streaming configuration.
dask_client: An existing distributed.Client to use. If None, a distributed.LocalCluster is created automatically.

Returns:

A new DaskEngine instance.

Examples

>>> from cudf_polars.engine.options import StreamingOptions
>>> opts = StreamingOptions(num_streaming_threads=4, fallback_mode="silent")
>>> with DaskEngine.from_options(opts) as engine:
...     result = pl.LazyFrame({"a": [1, 2, 3]}).collect(engine=engine)

gather_cluster_info() → list[ClusterInfo][source]#

Collect diagnostic information from every rank.

Returns:

List of ClusterInfo, one per rank.

gather_statistics( *, clear: bool = False, ) → list[Statistics][source]#

Collect statistics from every rank via client.run.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

List of Statistics, one per rank,
ordered by rank index.

global_statistics( *, clear: bool = False, ) → Statistics[source]#

Collect statistics from every rank and merge them into a single global statistics.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

A merged Statistics: per-stat counts
and values are summed, maxima are reduced with max. Formatters
are taken from rank 0.

property nranks: int[source]#

Number of ranks (for example GPUs or workers) in the cluster.

Local execution without a cluster returns 1.

Returns:

Number of ranks.

shutdown() → None[source]#

Shut down all Dask workers’ GPU resources.

Drains buffered Quent events from all workers before tearing down, then emits Engine.exit on the client.

If the cluster and client were created by this engine, they are also closed. Safe to call more than once. Must be called on the same thread that created the engine.

Raises:

ExceptionGroup: If one or more workers raise an unexpected exception during teardown.

class cudf_polars.engine.spmd.SPMDEngine( *, comm: Communicator | None = None, rapidsmpf_options: Options | None = None, executor_options: dict[str, Any] | None = None, engine_options: dict[str, Any] | None = None, )[source]#

Bases: StreamingEngine

Multi-GPU Polars engine for SPMD executions.

Bootstraps a RapidsMPF SPMD context and returns a matching engine.

SPMD execution model

SPMD (Single Program, Multiple Data) is a parallel programming model where each process runs the same Python script independently on its own slice of data. When launched with the RapidsMPF launcher rrun, multiple identical processes are started. Each process owns a rank-local LazyFrame representing its partition of the distributed dataset. Collective operations, such as shuffles, all-gathers, and joins, coordinate across ranks to produce a globally consistent result.

This class is the primary entry point for SPMD execution. It:

Bootstraps a communicator connecting all ranks. When launched with rrun this is a full UCXX communicator. When running as a normal single Python process (no rrun) it falls back to a lightweight single-rank communicator that requires no external communication library (no UCXX, Ray, or Dask).
Creates a RapidsMPF Context that owns GPU memory and a CUDA-stream pool.

All resources (communicator, stream pool, thread-pool) are released when shutdown() is called or the engine is used as a context manager.

Memory resource

SPMDEngine captures the configured device memory resource at construction and hands it to the RapidsMPF Context, which wraps it in an internal tracking RmmResourceAdaptor (exposed via BufferResource.device_mr_adaptor()). That tracking adaptor is installed as the current device resource so libcudf temporary allocations and the RapidsMPF Context share the same resource; the previous current resource is restored on shutdown.

To use a custom allocator, call rmm.mr.set_current_device_resource(your_mr) before constructing SPMDEngine. Do not pre-wrap it in RmmResourceAdaptor.

import rmm

# Optional: install a pool allocator before constructing SPMDEngine.
# rmm.mr.set_current_device_resource(
#     rmm.mr.PoolMemoryResource(rmm.mr.CudaMemoryResource())
# )
with SPMDEngine(...) as engine:
    ...

DataFrame and LazyFrame semantics

Because every rank runs an independent Python process, a DataFrame is always rank-local i.e. it contains only that rank’s partition of the distributed dataset. This is true whether the DataFrame originates from a file reader or from Python literals.

File-based sources (scan_parquet, scan_csv, …) distribute their work automatically: the engine assigns disjoint file- or row-group ranges to each rank, so different ranks produce different data.

An in-memory DataFrame (or one produced by a previous collect) is already rank-local by construction. Each rank processes its own copy in full; the engine does not re-slice it across ranks. In particular, the two patterns below are equivalent:

# One-step: scan and transform in a single pipeline
result = pl.scan_parquet(...).pipe(transform).collect(engine=engine)

# Two-step: collect an intermediate result, then transform
intermediate = pl.scan_parquet(...).collect(engine=engine)
result = intermediate.lazy().pipe(transform).collect(engine=engine)

In both cases rank k operates on exactly the data it read from parquet. The intermediate collect simply materializes the data in memory; it does not change which rows belong to which rank.

Query symmetry requirement

Every rank must issue the same sequence of Polars queries in the same order. Collective operations (shuffles, all-gathers, joins) are matched across ranks by a monotonically increasing operation ID; if one rank calls a collective that another rank does not, all ranks will deadlock. This means your driver script must be fully deterministic: avoid rank-conditional collect calls, early exits, or any branching that would cause different ranks to execute different query graphs.

Parameters:

comm: An already-bootstrapped communicator. When provided, the bootstrap step is skipped and the caller retains ownership; the communicator is not closed on shutdown. Pass this to share a single communicator across multiple engine lifetimes (e.g. a session-scoped pytest fixture). When None (default) a new communicator is bootstrapped automatically.
rapidsmpf_options: RapidsMPF-specific options. Defaults to the reading RAPIDSMPF_* environment variables.
executor_options: Executor-specific options (e.g. max_rows_per_partition).
engine_options: Engine-specific keyword arguments (e.g. raise_on_fail, parquet_options).

Attributes

`comm`	The active RapidsMPF communicator.
`context`	The active RapidsMPF streaming context.
`nranks`	Number of ranks (for example GPUs or workers) in the cluster.
`rank`	Rank index within the cluster (zero-based).

Methods

`from_options`(options)	Create an `SPMDEngine` from a `StreamingOptions` object.
`gather_cluster_info`()	Collect diagnostic information from every rank.
`gather_statistics`(*[, clear])	Collect statistics from every rank via an all-gather.
`global_statistics`(*[, clear])	Collect statistics from every rank and merge them into a single global statistics.
`shutdown`()	Shut down the engine and release all owned resources.

Raises:

TypeError: If executor_options or engine_options contains a reserved key.

Notes

Calls bind_to_gpu() at construction time, before RMM and communicator initialisation, so that CPU affinity, NUMA memory policy, and UCX_NET_DEVICES are set as early as possible. By default, binding is skipped under rrun (which already performs its own binding), see HardwareBindingPolicy.skip_under_rrun.

If a bootstrapped communicator is provided, the attached statistics object is used for all statistics logging, and enabled/disabled according to any options provided in rapidsmpf_options.

Examples

Context-manager style (recommended for scripts):

>>> with SPMDEngine() as engine:
...     result = (
...         df.lazy().group_by("a").agg(pl.col("b").sum()).collect(engine=engine)
...     )
...     full = allgather_polars_dataframe(engine=engine, local_df=result, op_id=0)

Direct style (Jupyter / long-lived clusters):

>>> engine = SPMDEngine()
>>> result = df.lazy().collect(engine=engine)
>>> engine.shutdown()

property comm: Communicator[source]#

The active RapidsMPF communicator.

Returns:

Active communicator.

Raises:

RuntimeError: If called after shutdown().

property context: Context[source]#

The active RapidsMPF streaming context.

Returns:

Active streaming context.

Raises:

RuntimeError: If called after shutdown().

classmethod from_options(options: StreamingOptions) → SPMDEngine[source]#

Create an SPMDEngine from a StreamingOptions object.

This is the recommended way to construct an SPMDEngine for typical use. All RapidsMPF, executor, and engine options are read from options; unset fields fall back to environment variables and then to built-in defaults.

Parameters:

options: Unified streaming configuration.

Returns:

A new SPMDEngine instance.

Examples

>>> from cudf_polars.engine.options import StreamingOptions
>>> opts = StreamingOptions(num_streaming_threads=8, fallback_mode="silent")
>>> with SPMDEngine.from_options(opts) as engine:
...     result = df.lazy().collect(engine=engine)

gather_cluster_info() → list[ClusterInfo][source]#

Collect diagnostic information from every rank.

This is a collective operation, every rank must call it.

Returns:

List of ClusterInfo, one per rank.

gather_statistics( *, clear: bool = False, ) → list[Statistics][source]#

Collect statistics from every rank via an all-gather.

This is a collective operation, every rank must call it.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

List of Statistics, one per rank,
ordered by rank index.

global_statistics( *, clear: bool = False, ) → Statistics[source]#

Collect statistics from every rank and merge them into a single global statistics.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

A merged Statistics: per-stat counts
and values are summed, maxima are reduced with max. Formatters
are taken from rank 0.

property nranks: int[source]#

Number of ranks (for example GPUs or workers) in the cluster.

Local execution without a cluster returns 1.

Returns:

Number of ranks.

property rank: int[source]#

Rank index within the cluster (zero-based).

Returns:

Rank index.

Raises:

RuntimeError: If called after shutdown().

shutdown() → None[source]#

Shut down the engine and release all owned resources.

Idempotent: safe to call more than once. Must be called on the same thread that created the engine.

class cudf_polars.engine.default_singleton_engine.DefaultSingletonEngine(**kwargs: Any)[source]#

Bases: SPMDEngine

Process-wide single-GPU singleton specialization of SPMDEngine.

At most one live instance exists per process. Use get_or_create() to obtain it and shutdown() to tear it down.

Always constructs a single-rank communicator and uses default RapidsMPF, executor, and engine settings from the environment.

Users needing custom configuration should construct an engine explicitly. See RayEngine, DaskEngine, and SPMDEngine.

Methods

`get_or_create`()	Return the live singleton, constructing one if needed.
`shutdown`()	Shut down the live default singleton, if any.

Examples

Constructed automatically when using engine="gpu":

>>> result = df.lazy().collect(engine="gpu")

Or constructed explicitly:

>>> engine = DefaultSingletonEngine.get_or_create()
>>> result = df.lazy().collect(engine=engine)

classmethod get_or_create() → DefaultSingletonEngine[source]#

Return the live singleton, constructing one if needed.

Construction runs on a dedicated worker thread so the rapidsmpf Context is born on the same thread that will eventually tear it down.

Raises:

RuntimeError: If any other StreamingEngine is currently alive.

static shutdown() → None[source]#

Shut down the live default singleton, if any. Idempotent.

Submits teardown to the dedicated worker thread, the same thread that constructed the rapidsmpf Context, and waits up to SHUTDOWN_TIMEOUT_SECONDS seconds.

The engine classes share a common base class:

class cudf_polars.engine.core.StreamingEngine( *, nranks: int, executor_options: dict[str, Any], engine_options: dict[str, Any], exit_stack: ExitStack | None = None, )[source]#

Bases: GPUEngine

Base class for multi-GPU Polars engines.

The engine manages the lifecycle of a streaming execution and can be used as a context manager. On exit, shutdown() is called.

Parameters:

nranks: Number of ranks (workers or GPUs) in the cluster.
executor_options: Executor-specific options (e.g. max_rows_per_partition).
engine_options: Engine-specific keyword arguments (e.g. raise_on_fail, parquet_options).
exit_stack: A contextlib.ExitStack whose registered contexts are closed when shutdown() is called. If None, an empty stack is created.

Attributes

nranks

Number of ranks (for example GPUs or workers) in the cluster.

Methods

`gather_cluster_info`()	Collect diagnostic information from every rank.
`gather_statistics`(*[, clear])	Collect statistics from every rank.
`global_statistics`(*[, clear])	Collect statistics from every rank and merge them into a single global statistics.
`shutdown`()	Shut down engine and release all owned resources.

Notes

The engine must be created and shut down on the same thread. In particular, destruction and context manager exit must occur on the thread that created the instance.

gather_cluster_info() → list[ClusterInfo][source]#

Collect diagnostic information from every rank.

Returns:

List of ClusterInfo, one per rank.

gather_statistics( *, clear: bool = False, ) → list[Statistics][source]#

Collect statistics from every rank.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

List of Statistics, one per rank,
ordered by rank index.

global_statistics( *, clear: bool = False, ) → Statistics[source]#

Collect statistics from every rank and merge them into a single global statistics.

Parameters:

clear: If True, clear each rank’s statistics after gathering.

Returns:

A merged Statistics: per-stat counts
and values are summed, maxima are reduced with max. Formatters
are taken from rank 0.

property nranks: int[source]#

Number of ranks (for example GPUs or workers) in the cluster.

Local execution without a cluster returns 1.

Returns:

Number of ranks.

shutdown() → None[source]#

Shut down engine and release all owned resources.

Idempotent: safe to call more than once. Must be called on the same thread that created the engine.

class cudf_polars.engine.core.ClusterInfo( pid: int, hostname: str, cuda_visible_devices: str | None, gpu_uuid: str, device_memory: int | None = None, )[source]#

Diagnostic information about a single rank in the cluster.

Attributes

pid	Process ID of the current rank.
hostname	Hostname of the machine running this rank.
cuda_visible_devices	Value of `CUDA_VISIBLE_DEVICES`, or `None` if unset.
gpu_uuid	UUID of the current CUDA device.
device_memory	Total device memory in bytes, or `None` if unknown.

Methods

local()

Build a ClusterInfo for the current process and GPU.

classmethod local() → ClusterInfo[source]#

Build a ClusterInfo for the current process and GPU.

Returns:

Diagnostic information for this rank.

Persisted results#

Returned by engine.execute() to keep query results GPU-resident (see Keeping results on the GPU with engine.execute()).

class cudf_polars.engine.persisted_result.PersistedQueryResult( backend: PersistedBackend, uid: str, query_id: UUID, ranks: list[int], schema: dict[str, DataType], )[source]#

Distributed query result whose partitions remain on their producing ranks.

Returned by engine.execute(). It must be collected or executed only with the engine that produced it, never with a different engine (including the default host Polars engine, which cannot read the GPU-resident partitions). Using another engine is unsupported and not currently guarded against.

Collection is one-shot. Once collected, a result cannot be collected again; attempting to do so raises an exception.

If never collected, the partitions are released when this result, and any LazyFrame derived from lazy(), is garbage-collected. They may also be released explicitly via release() or the context-manager protocol.

Parameters:

backend: Engine hook used to release persisted partitions.
uid: Store identifier used to locate the persisted partitions.
query_id: Identifier of the query that produced the partitions.
ranks: Ranks that produced a partition.
schema: Output schema as a {column_name: polars_dtype} mapping.

Methods

`lazy`()	Return a `LazyFrame` backed by the persisted partitions.
`release`()	Release the persisted partitions now (idempotent).

lazy() → LazyFrame[source]#

Return a LazyFrame backed by the persisted partitions.

Collecting with the producing engine runs the scan on each owning process, which moves its partition out of the local store. The result can be collected only once. Collecting the returned dataframe a second time raises.

Returns:

LazyFrame with one partition per original rank.

release() → None[source]#

Release the persisted partitions now (idempotent).

Invalidates any LazyFrame previously returned by lazy().

Configuration#

class cudf_polars.engine.options.StreamingOptions( num_streaming_threads: int | Unspecified = <factory>, num_streams: int | Unspecified = <factory>, log: Literal['NONE', 'PRINT', 'WARN', 'INFO', 'DEBUG', 'TRACE'] | Unspecified = <factory>, statistics: bool | Unspecified = <factory>, memory_reserve_timeout: str | Unspecified = <factory>, allow_overbooking_by_default: bool | Unspecified = <factory>, pinned_memory: bool | Unspecified = <factory>, pinned_initial_pool_size: int | Unspecified = <factory>, pinned_max_pool_size: str | Unspecified = <factory>, spill_device_limit: str | Unspecified = <factory>, periodic_spill_check: str | Unspecified = <factory>, unbounded_file_read_cache: str | Unspecified = <factory>, num_py_executors: int | Unspecified = <factory>, fallback_mode: str | Unspecified = <factory>, max_rows_per_partition: int | Unspecified = <factory>, broadcast_limit: int | Unspecified = <factory>, target_partition_size: int | Unspecified = <factory>, dynamic_planning: dict[str, Any] | DynamicPlanningOptions | None | Unspecified = <factory>, join_filter_pushdown: dict[str, Any] | JoinFilterPushdownOptions | None | Unspecified = <factory>, sink_to_directory: bool | Unspecified = <factory>, quent_context: QuentContext | None | Unspecified = <factory>, raise_on_fail: bool | Unspecified = <factory>, parquet_options: dict[str, Any] | ParquetOptions | Unspecified = <factory>, memory_resource_config: MemoryResourceConfig | Unspecified = <factory>, hardware_binding: HardwareBindingPolicy | Unspecified = <factory>, allow_gpu_sharing: bool | Unspecified = <factory>, )[source]#

High-level configuration for the cudf-polars streaming executor and RapidsMPF.

Options are grouped into three categories:

RapidsMPF: runtime and memory behavior (e.g. spilling, threading).
Executor: query execution and partitioning behavior.
Engine: Polars integration, IO configuration, hardware binding.

All fields default to UNSPECIFIED and follow this precedence:

Explicit value.
Environment variable.
Built-in default.

Parameters:

num_streaming_threads: Threads used to execute coroutines. Env: RAPIDSMPF_NUM_STREAMING_THREADS. Default: 1. Category: rapidsmpf.
num_streams: CUDA streams for concurrent GPU execution. Env: RAPIDSMPF_NUM_STREAMS. Default: 16. Category: rapidsmpf.
log: Log level ("NONE", "PRINT", "WARN", "INFO", "DEBUG", "TRACE"). Env: RAPIDSMPF_LOG. Default: "WARN". Category: rapidsmpf.
statistics: Enable performance metrics. Env: RAPIDSMPF_STATISTICS. Default: False. Category: rapidsmpf.
memory_reserve_timeout: Timeout for memory reservations (e.g. "100ms"). Env: RAPIDSMPF_MEMORY_RESERVE_TIMEOUT. Default: "100ms". Category: rapidsmpf.
allow_overbooking_by_default: Allow overallocation in reservation APIs. Env: RAPIDSMPF_ALLOW_OVERBOOKING_BY_DEFAULT. Default: True. Category: rapidsmpf.
pinned_memory: Enable pinned host memory. Env: RAPIDSMPF_PINNED_MEMORY. Default: False. Category: rapidsmpf.
pinned_initial_pool_size: Initial pinned memory pool size (bytes). Env: RAPIDSMPF_PINNED_INITIAL_POOL_SIZE. Default: 0. Category: rapidsmpf.
pinned_max_pool_size: Maximum pinned host memory pool size (e.g. "4GiB", "50%"). Env: RAPIDSMPF_PINNED_MAX_POOL_SIZE. Default: 80% of per-GPU host memory. Category: rapidsmpf.
spill_device_limit: Device memory soft limit before spilling (e.g. "80%" or bytes). Env: RAPIDSMPF_SPILL_DEVICE_LIMIT. Default: "80%". Category: rapidsmpf.
periodic_spill_check: Interval between spill checks (e.g. "1ms"). Env: RAPIDSMPF_PERIODIC_SPILL_CHECK. Default: "1ms". Category: rapidsmpf.
unbounded_file_read_cache: Cache file-read results in the Context’s message storage. Accepts a memory type ("host", "pinned", "device") or "disabled". Primarily for benchmarking. Each file slice must be read with identical parameters (see rapidsmpf docs). Env: RAPIDSMPF_UNBOUNDED_FILE_READ_CACHE. Default: "disabled". Category: rapidsmpf.
num_py_executors: Workers for the internal Python ThreadPoolExecutor. Env: CUDF_POLARS__EXECUTOR__NUM_PY_EXECUTORS. Default: 8. Category: executor.
fallback_mode: Fallback behavior ("warn", "raise", "silent"). Env: CUDF_POLARS__EXECUTOR__FALLBACK_MODE. Default: "warn". Category: executor.
max_rows_per_partition: Maximum rows per partition. Env: CUDF_POLARS__EXECUTOR__MAX_ROWS_PER_PARTITION. Default: 1_000_000. Category: executor.
broadcast_limit: Maximum byte size for broadcast joins. Env: CUDF_POLARS__EXECUTOR__BROADCAST_LIMIT. Default: "auto". Category: executor.
target_partition_size: Target IO partition size (bytes). 0 = auto. Env: CUDF_POLARS__EXECUTOR__TARGET_PARTITION_SIZE. Default: auto. Category: executor.
dynamic_planning: Dynamic planning config, dict or DynamicPlanningOptions. None disables. Env: CUDF_POLARS__EXECUTOR__DYNAMIC_PLANNING. Default: enabled. Category: executor.
join_filter_pushdown: Config for join filter pushdown optimizations, dict or JoinFilterPushdownOptions. None disables the rewrite. Env: CUDF_POLARS__EXECUTOR__JOIN_FILTER_PUSHDOWN and CUDF_POLARS__EXECUTOR__JOIN_FILTER_PUSHDOWN__*. Default: enabled. Category: executor.
sink_to_directory: Whether multi-partition sink operations should write to a directory rather than a single file. The spmd/ray/dask engines always use True; passing False raises ValueError. Env: CUDF_POLARS__EXECUTOR__SINK_TO_DIRECTORY. Default: True (forced by the streaming engines). Category: executor.
quent_context: Quent tracing context, or None to disable tracing. Env: CUDF_POLARS__EXECUTOR__QUENT_CONTEXT (true/false). Default: None (disabled).
raise_on_fail: Raise instead of falling back to CPU. Default: False. Category: engine.
parquet_options: Parquet configuration, dict or ParquetOptions. Env: CUDF_POLARS__PARQUET_OPTIONS__*. Category: engine.
memory_resource_config: RMM configuration, dict or MemoryResourceConfig. Env: CUDF_POLARS__MEMORY_RESOURCE_CONFIG__*. Category: engine.
hardware_binding: Hardware binding policy. Pass a HardwareBindingPolicy instance for fine-grained control. Env: CUDF_POLARS__HARDWARE_BINDING (JSON object, e.g. '{"enabled": false}'). Default: HardwareBindingPolicy(). Category: engine.
allow_gpu_sharing: When False (default), the engine raises if multiple ranks share the same physical GPU. Env: CUDF_POLARS__ALLOW_GPU_SHARING. Default: False. Category: engine.

Methods

`from_dict`(data)	Build a `StreamingOptions` from a plain dictionary.
`to_dict`()	Return all explicitly-set fields as a plain dictionary.
`to_engine_options`()	Build a `pl.GPUEngine` kwargs dict from the engine fields.
`to_executor_options`()	Build a `StreamingExecutor` kwargs dict from the executor fields.
`to_rapidsmpf_options`()	Build a `rapidsmpf.config.Options` from the RapidsMPF fields.

Examples

>>> StreamingOptions(num_streaming_threads=8, log="DEBUG", fallback_mode="silent")
StreamingOptions(...)

classmethod from_dict( data: dict[str, Any], ) → StreamingOptions[source]#

Build a StreamingOptions from a plain dictionary.

Keys must be field names of StreamingOptions. Values of None and missing keys both leave the corresponding field at UNSPECIFIED.

Parameters:

data: Flat dictionary of option name to value. Unknown keys raise TypeError.

Returns:

A new StreamingOptions instance.

Examples

>>> StreamingOptions.from_dict(
...     {"fallback_mode": "silent", "num_streaming_threads": 4}
... )
StreamingOptions(...)
>>> StreamingOptions.from_dict({})  # all fields UNSPECIFIED
StreamingOptions(...)

to_dict() → dict[str, Any][source]#

Return all explicitly-set fields as a plain dictionary.

Fields that are UNSPECIFIED are omitted. The result can be round-tripped via from_dict().

Returns:

Mapping of field name to value for every non-UNSPECIFIED field.

Examples

>>> StreamingOptions(fallback_mode="silent").to_dict()
{'fallback_mode': 'silent'}
>>> StreamingOptions.from_dict(
...     StreamingOptions(fallback_mode="silent").to_dict()
... )
StreamingOptions(...)

to_engine_options() → dict[str, Any][source]#

Build a pl.GPUEngine kwargs dict from the engine fields.

Only fields that are not UNSPECIFIED are included. ConfigOptions.from_polars_engine handles environment variables for any omitted fields.

to_executor_options() → dict[str, Any][source]#

Build a StreamingExecutor kwargs dict from the executor fields.

Only fields that are not UNSPECIFIED are included. StreamingExecutor reads CUDF_POLARS__EXECUTOR__* environment variables for any omitted fields.

to_rapidsmpf_options() → Options[source]#

Build a rapidsmpf.config.Options from the RapidsMPF fields.

RAPIDSMPF_* environment variables are resolved at StreamingOptions construction time, so any field still UNSPECIFIED here has no environment variable and no explicit value; the rapidsmpf C++ library will apply its own built-in default for those.

cudf_polars.engine.options.UNSPECIFIED = UNSPECIFIED#

Singleton sentinel for all StreamingOptions fields.

A field set to UNSPECIFIED after construction means no explicit value and no matching environment variable was found; the underlying library will apply its own built-in default.

class cudf_polars.engine.hardware_binding.HardwareBindingPolicy( skip_under_rrun: bool = True, enabled: bool = True, enable_once: bool = True, raise_on_fail: bool = False, cpu: bool = True, memory: bool = True, network: bool = False, )[source]#

Policy controlling topology-aware hardware binding.

Determines whether rapidsmpf.rrun.rrun.bind() is invoked to pin the calling process to CPU cores, NUMA memory nodes, and network devices local to the worker’s GPU.

The GPU to bind to is resolved from CUDA_VISIBLE_DEVICES. Each frontend is responsible for setting this variable per worker (Dask via the nanny preload or SpecCluster, Ray via num_gpus=1 scheduling, SPMD via rrun). If CUDA_VISIBLE_DEVICES is unset, binding falls back to GPU 0.

The default instance (HardwareBindingPolicy()) enables binding once per process with soft failure handling.

Parameters:

skip_under_rrun: When True (the default), binding is skipped if the process was launched via rrun, because rrun already performs hardware binding at launch time. If binding is skipped, all other options are ignored.
enabled: Whether binding is enabled. False disables all binding.
enable_once: When True, binding is performed at most once per process; subsequent calls to bind_to_gpu() are no-ops. When False, binding is attempted on every call.
raise_on_fail: When True, binding failures (e.g. CPU affinity, NUMA memory policy, or topology discovery) raise an exception. When False (the default), failures are silently ignored.
cpu: Whether to bind CPU cores. Enabled by default.
memory: Whether to bind NUMA memory nodes. Enabled by default.
network: Whether to bind network devices. Disabled by default because UCX is usually capable of automatically determining affinity to the appropriate NICs, and on certain systems a more complex binding is necessary to avoid network-affinity problems.

cudf_polars.engine.hardware_binding.bind_to_gpu( policy: HardwareBindingPolicy, ) → None[source]#

Bind the calling process to resources topologically close to a GPU.

Thread-safe wrapper around rapidsmpf.rrun.rrun.bind() governed by policy.

When policy.enable_once is True (the default), double-checked locking with a module-level lock guarantees that the underlying bind() is called at most once per process, regardless of how many frontend engines are constructed or from how many threads.

Parameters:

policy: The HardwareBindingPolicy controlling binding behavior.

SPMD helpers#

cudf_polars.engine.spmd.allgather_polars_dataframe( *, engine: SPMDEngine, local_df: pl.DataFrame, op_id: int, ) → pl.DataFrame[source]#

AllGather a rank-local DataFrame so every rank receives the full result.

Each rank contributes its local local_df partition and receives the concatenation of all ranks’ partitions in rank order. This is the SPMD equivalent of a distributed collect: after the call, every rank holds the same complete dataset.

Parameters:

engine: The active SPMDEngine.
local_df: Rank-local DataFrame to contribute.
op_id: Operation ID for this AllGather collective. Must be identical on every rank. For example, use reserve_op_id() to obtain a collision-free ID from the same pool used internally by cudf-polars. Avoid passing hardcoded integers.

Returns:

DataFrame containing rows from all ranks, ordered by rank.

Raises:

RuntimeError: If engine has already been shut down.

cudf_polars.streaming.actor_graph.collectives.common.reserve_op_id() → Iterator[int][source]#

Reserve a single collective operation ID.

This function and the ID it yields must only be used outside of a run_actor_graph call. It is intended for SPMD mode, where operations such as gathering results across ranks are performed directly rather than through the actor graph. The contained block _must_ wait for completion of the collective.

Yields:

collective_idint: A vacant collective ID reserved from the global vacancy pool.

Internal configuration objects#

These dataclasses back the engine_options surfaced by pl.GPUEngine and StreamingOptions. Most users interact with them through StreamingOptions fields rather than directly.

Configuration utilities for the cudf-polars engine.

Most users will not construct these objects directly. Instead, you’ll pass keyword arguments to GPUEngine. The majority of the options are passed as **kwargs and collected into the configuration described below:

>>> import polars as pl
>>> engine = pl.GPUEngine(
...     executor="streaming",
...     executor_options={"fallback_mode": "raise"}
... )

class cudf_polars.utils.config.DynamicPlanningOptions( sample_chunk_count: int = <factory>, join_prefilter_threshold: float = <factory>, join_prefilter_max_key_columns: int | None = <factory>, join_prefilter_trace: bool = <factory>, )[source]#

Configuration for dynamic shuffle planning.

When enabled, shuffle decisions for GroupBy/Join/Unique operations are made at runtime by sampling real chunks.

To enable dynamic planning, pass a DynamicPlanningOptions instance to StreamingExecutor(dynamic_planning=...). To disable it, pass None (the default).

These options can be configured via environment variables with the prefix CUDF_POLARS__EXECUTOR__DYNAMIC_PLANNING__.

Parameters:

sample_chunk_count: The maximum number of chunks to sample before making dynamic-planning decisions. Default is 2.
join_prefilter_threshold: Row-count ratio (small / large) below which one side of a join is filtered by a bloom filter built from the other side before performing the join. Set to 0 to disable. Default is 0.5.
join_prefilter_max_key_columns: Maximum number of columns from the join-key prefix to use for the prefilter. Set to None to use the full join-key list. Default is 1.
join_prefilter_trace: Whether to collect input/output row counts around applied join prefilters. Default is False.

class cudf_polars.utils.config.JoinFilterPushdownOptions( threshold: float = <factory>, trace: bool = <factory>, )[source]#

Configuration options for join filter pushdown in the logical plan.

When performing a join between two tables, it is often favourable to pre-filter one side of the join with the keys (full or partial) of the other side. This can reduce the size of tables that actually participate in the join.

cudf-polars supports a form of this where we can rewrite inner joins by selecting a side to be filtered by the keys of the other side.

Pass None to StreamingExecutor(join_filter_pushdown=...) to disable the rewrite.

These options can be configured via environment variables with the prefix CUDF_POLARS__EXECUTOR__JOIN_FILTER_PUSHDOWN__.

Parameters:

threshold: Row-count ratio (key-provider-rows / to-be-filtered-table-rows) below which a filter on is inserted on the to-be-filtered table. Default is 0.5.
trace: Whether to emit plan-time trace decisions for filter decisions. Default is False.

class cudf_polars.utils.config.MemoryResourceConfig( qualname: str = <factory>, options: dict[str, ~typing.Any] | None = <factory>, )[source]#

Configuration for the default memory resource.

Parameters:

qualname: The fully qualified name of the memory resource class to use.
options: This can be either a dictionary representing the options to pass to the memory resource class, or, a dictionary representing a nested memory resource configuration. The presence of “qualname” field indicates a nested memory resource configuration.

Examples

Create a memory resource config for a single memory resource:

>>> MemoryResourceConfig(
...     qualname="rmm.mr.CudaAsyncMemoryResource",
...     options={"initial_pool_size": 100},
... )

Create a memory resource config for a nested memory resource configuration:

>>> MemoryResourceConfig(
...     qualname="rmm.mr.PrefetchResourceAdaptor",
...     options={
...         "upstream_mr": {
...             "qualname": "rmm.mr.PoolMemoryResource",
...             "options": {
...                 "upstream_mr": {
...                     "qualname": "rmm.mr.ManagedMemoryResource",
...                 },
...                 "initial_pool_size": 256,
...             },
...         }
...     },
... )

create_memory_resource() → rmm.mr.DeviceMemoryResource[source]#: Create a memory resource from the configuration.

classmethod default() → MemoryResourceConfig[source]#

The default memory resource config.

This defaults to a CUDA Async Memory Resource with

No initial pool size
A release threshold equal to 90% of the size of the device’s memory.

class cudf_polars.utils.config.ParquetOptions( chunked: bool = <factory>, n_output_chunks: int = <factory>, chunk_read_limit: int = <factory>, pass_read_limit: int = <factory>, max_footer_samples: int = <factory>, max_row_group_samples: int = <factory>, use_rapidsmpf_native: bool = <factory>, prefetch_file_metadata: bool = <factory>, use_jit_filter: bool = <factory>, )[source]#

Configuration for the cudf-polars Parquet engine.

These options can be configured via environment variables with the prefix CUDF_POLARS__PARQUET_OPTIONS__.

Parameters:

chunked

Whether to use libcudf’s ChunkedParquetReader or ChunkedParquetWriter to read/write the parquet dataset in chunks. This is useful when reading/writing very large parquet files.

n_output_chunks

Split the dataframe in n_output_chunks when using libcudf’s ChunkedParquetWriter.

chunk_read_limit

Limit on total number of bytes to be returned per read, or 0 if there is no limit.

pass_read_limit

Limit on the amount of memory used for reading and decompressing data or 0 if there is no limit.

max_footer_samples

Maximum number of file footers to sample for metadata. This option is currently used by the streaming executor to gather datasource statistics before generating a physical plan. Set to 0 to avoid metadata sampling. Default is 3.

max_row_group_samples

Maximum number of row-groups to sample for unique-value statistics. This option may be used by the streaming executor to optimize the physical plan. Default is 1.

Set to 0 to avoid row-group sampling. Note that row-group sampling will also be skipped if max_footer_samples is 0.

use_rapidsmpf_native

Whether to use the native rapidsmpf node for parquet reading. This option is only used by the streaming executor. Default is False.

prefetch_file_metadata

Whether to prefetch parquet file metadata and pass it through parquet_metadatas to avoid rereading file footers.

use_jit_filter

Whether to use JIT compilation for post-read filtering in Parquet scans. When enabled, filter predicates are JIT-compiled to CUDA kernels for improved performance on large datasets with complex filters. Default is False.

class cudf_polars.utils.config.StreamingExecutor( cluster: Cluster | None = <factory>, fallback_mode: StreamingFallbackMode = <factory>, max_rows_per_partition: int = <factory>, target_partition_size: int = <factory>, broadcast_limit: int = <factory>, client_device_threshold: float = <factory>, sink_to_directory: bool | None = <factory>, dynamic_planning: DynamicPlanningOptions | None = <factory>, join_filter_pushdown: JoinFilterPushdownOptions | None = <factory>, max_io_threads: int = <factory>, num_py_executors: int = <factory>, min_device_size: int | None = None, spmd_context: SPMDContext | None = None, ray_context: RayContext | None = None, dask_context: DaskContext | None = None, quent_context: QuentContext | None = <factory>, )[source]#

Configuration for the cudf-polars streaming executor.

These options can be configured via environment variables with the prefix CUDF_POLARS__EXECUTOR__.

Parameters:

cluster

The cluster configuration for the streaming executor. Cluster.DEFAULT_SINGLETON by default.

Cluster.DEFAULT_SINGLETON: Single-GPU execution
Cluster.SPMD: Multi-GPU SPMD execution
Cluster.RAY: Multi-GPU Ray execution
Cluster.DASK: Multi-GPU Dask execution

fallback_mode

How to handle errors when the GPU engine fails to execute a query. StreamingFallbackMode.WARN by default.

This can be set using the CUDF_POLARS__EXECUTOR__FALLBACK_MODE environment variable.

max_rows_per_partition

The maximum number of rows to process per partition. 1_000_000 by default. When the number of rows exceeds this value, the query will be split into multiple partitions and executed in parallel.

target_partition_size

Target partition size, in bytes, for IO tasks. This configuration currently controls how large parquet files are split into multiple partitions. Files larger than target_partition_size bytes are split into multiple partitions.

This can be set via

keyword argument to polars.GPUEngine
the CUDF_POLARS__EXECUTOR__TARGET_PARTITION_SIZE environment variable

By default, cudf-polars uses the minimum of 1.5GB or 2.5% of the minimum device size in the cluster. If pynvml cannot query the the device size(s), the default target_partition_size will be 1.5GB.

broadcast_limit

The maximum number of bytes to broadcast in a single operation. By default, cudf-polars uses the minimum of 16GB or 15% of the minimum device size in the cluster. If pynvml cannot query the the device size(s), the default broadcast_limit will be 16GB.

client_device_threshold

Threshold for spilling data from device memory. Default is 50% of device memory on the client process.

sink_to_directory

Whether multi-partition sink operations write to a directory rather than a single file. For the spmd, ray, and dask clusters this is always True; setting it to False raises a ValueError.

dynamic_planning

Options controlling dynamic shuffle planning. See DynamicPlanningOptions for more.

join_filter_pushdown

Options controlling the logical join-domain prefilter rewrite. See JoinFilterPushdownOptions for more. None disables the rewrite.

Enable through environment variables with CUDF_POLARS__EXECUTOR__JOIN_FILTER_PUSHDOWN=1.

max_io_threads

Maximum number of IO threads. Default is 4. This controls the parallelism of IO operations when reading data.

num_py_executors

Maximum number of workers for the Python ThreadPoolExecutor. Default is 8.

quent_context

Quent tracing context. When None (default), Quent tracing is disabled. Pass a QuentContext instance to enable tracing. Can be set via the CUDF_POLARS__EXECUTOR__QUENT_CONTEXT environment variable (true enables tracing with a default context, false disables it).

Notes

The streaming executor does not currently support profiling a query via the .profile() method. We recommend using nsys to profile queries.

drop_unserializable() → StreamingExecutor[source]#

Return a copy without the per-cluster contexts that cannot be pickled.

The streaming executor holds live, process-local handles (communicators, streaming contexts, thread pools) that must not be shipped to a worker/actor.

Returns:

A copy of this executor with the cluster contexts set to None.

class cudf_polars.utils.config.StreamingFallbackMode(*values)[source]#

How the streaming executor handles operations that don’t support multiple partitions.

Upon encountering an unsupported operation, the streaming executor will fall back to using a single partition, which might use a large amount of memory.

StreamingFallbackMode.WARN : Emit a warning and fall back to a single partition.
StreamingFallbackMode.SILENT: Silently fall back to a single partition.
StreamingFallbackMode.RAISE : Raise an exception.

Quent Integration#

cudf-polars can emit Quent events, which can be used to profile your queries.

Quent telemetry tracing.

class cudf_polars.quent.Attribute( name: str, value: int | float | str | bool | list[int] | list[float] | list[str] | list[bool] | dict[str, int | float | str | bool | list[int] | list[float] | list[str] | list[bool] | dict[str, Value]] | None, )[source]#: A Quent custom attribute.

class cudf_polars.quent.Engine( id: ~uuid.UUID = <factory>, implementation: ~cudf_polars.quent._types.Implementation = <factory>, )[source]#: A Quent Engine.

class cudf_polars.quent.Implementation( name: str = 'cudf-polars', version: str = '26.10.00a156', custom_attributes: list[~cudf_polars.quent._types.Attribute] = <factory>, )[source]#

Engine implementation metadata.

to_dict() → dict[str, Any][source]#: Serialize to a plain dict for JSON output.

class cudf_polars.quent.QuentContext( *, engine: ~cudf_polars.quent._types.Engine = <factory>, query_group: ~cudf_polars.quent._types.QueryGroup = <factory>, query: ~cudf_polars.quent._types.Query = <factory>, )[source]#

A Quent context that is globally valid for a query.

Parameters:

engine

A Quent Engine object. By default, a new Engine object is created with the cudf-polars Implementation.

query_group

A Quent QueryGroup object. By default, a new QueryGroup with no instance name is created.

This query group is used for all queries executed by this engine.

query

A Query Query object. By default, a new Query with no instance name is created.

classmethod deserialize(data: bytes) → Self[source]#

Deserialize a QuentContext from bytes.