Polars Cloud client 0.11.0

🏆 Highlights

Collect distributed results straight back to the client

Distributed queries can now return their results to your machine, without a
sink in between. collect() gives you a single DataFrame, collect_batches() gives you an
iterator of them, in case the result is larger:

import polars as pl
import polars_cloud as pc

ctx = pc.ClusterContext("https://my-scheduler:5051")

df = pl.scan_parquet("s3://bucket/data/*.parquet").remote(ctx).collect()

for batch in pl.scan_parquet("s3://bucket/data/*.parquet").remote(ctx).collect_batches():
    handle(batch)

collect() preserves row order. collect_batches() doesn't by default, so pass
maintain_order=True if you need it. Both take a ttl for how long the result stays
available on the cluster, and both can be called straight off lf.remote(ctx) or after
picking an execution mode with .distributed() or .single_node().

Results are streamed to the client, so this only works on a ClusterContext.

This sits next to sink_batches, which hands each batch to a callback running on the workers.
Use collect_batches when you want the data on the client ; use sink_batches when the work
should stay on the cluster.

⚠️ Breaking change

Workspace.delete() no longer removes your cloud infrastructure.

It used to delete the workspace and take the CloudFormation stack with it. It now only deletes
the workspace, and refuses to run while a provider is still attached. Tearing down the AWS
resources is a separate step:

# before 0.11.0
ws.delete()             # removed the workspace and the CloudFormation stack

# 0.11.0
ws.aws.disconnect()     # detaches AWS and prints the stack you need to delete
ws.delete()             # then remove the workspace

Note that disconnect() does not delete the stack for you either. It removes the connection
and opens the CloudFormation console so you can delete the stack yourself.

If you have scripts that call ws.delete() to clean up, they will now leave AWS resources
running. Add the disconnect() call and the manual stack deletion.

Workspaces connect to an infrastructure provider

Creating a workspace and attaching cloud infrastructure to it are now two separate steps. A
workspace starts empty, then you connect a provider to it:

ws = pc.Workspace.create("my-workspace", "my-org")
ws.connect_provider()             # AWS by default; opens the CloudFormation flow
ws.aws.is_connected()

connect_provider takes a ProviderType (AWS today) and a verify flag. Everything about
the AWS side of a workspace can be found under ws.aws: is_connected(), connect(),
disconnect() and wait_until_connected().

The same split is in the CLI:

pc workspace create my-workspace --connect-aws
pc workspace aws connect
pc workspace aws verify
pc workspace aws disconnect

This replaces Workspace.setup, Workspace.deploy and the workspace status API, all of which
are deprecated (see below). The reason for the change is to make room for providers other than AWS.

✨ Enhancements

Distributed expressions

• Cumulative aggregations run distributed: cum_sum, cum_count, cum_min, cum_max and
  cum_prod, forward and with reverse=True. Each partition scans locally and is combined with the
  offset of everything outside it, so nulls, NaNs and empty partitions match a single-node run.
• is_empty and has_nulls are distributed, both as group-by aggregations and over a whole frame.
• Fixed-window rolling expressions (rolling_min, rolling_max, rolling_mean, ...) run
  distributed, including center=True. They lower to an overlapped partitioning, which costs a
  local overlap exchange instead of a full shuffle. The rolling_*_by variants still run
  single-node.

Autoscaling for K8s

Clusters can grow and shrink their worker pool while running, instead of holding a fixed number of
workers. See Autoscaling.

• A query sizes itself with .distributed(min_workers=..., max_workers=...). min_workers is what
  must be available before execution starts; the cluster scales up to meet it and the query waits.
  max_workers caps the query and sets its partition count.
• Polars does not create machines itself. It delegates to a scaling service you run, over REST
  (GET /scale_config, POST /scale_to) or gRPC (ScalingService), so scaling goes through
  whatever already provisions your machines. The Kubernetes operator ships with its own scaler.
• The scheduler's leader tracks demand as the sum of the worker counts of all running queries, asks
  your service to match it, and never scales below the workers that currently hold a query. It
  scales down after 60 seconds of idle.
• scheduler.default_workers_per_query sets the size of a query that does not ask for one, and is
  required when [scaling] is configured. n_workers is deprecated in favour of it and
  max_workers_per_query.
• scale_up_timeout (default 5 minutes) is the window your service has to provision workers, and
  accepts ISO 8601 or jiff friendly durations.
• scale_down_on_idle_only restricts scale-down to an idle cluster, for services that cannot honor
  workers_to_keep.

Query labels work in direct mode

Labels passed on a ClusterContext query used to be dropped silently. They're now sent
through, and validated on the client before a cluster is started, so a bad label fails fast
instead of after the compute is up:

lf.remote(ctx).labels(["nightly", "etl"]).sink_parquet("s3://bucket/out.parquet")

A query takes at most 64 labels. Names have to be non-empty, at most 32 characters, free of
control characters and without surrounding whitespace.

Results come back sooner

DirectQuery.await_result() and await_result_async() now make one blocking call that the
server holds open, instead of polling in a loop with backoff. You get the result as soon as
it's ready rather than on the next poll tick, and the status chatter is gone.

Hive-partitioned scans on multiple partition columns no longer introduce a shuffle either.

Opt-in OpenTelemetry tracing

The client can emit OTLP traces for authentication, context setup and query submission.
Nothing is exported unless you set one of OTLP_ENDPOINT, OTEL_EXPORTER_OTLP_ENDPOINT or
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT. There is no telemetry by default.

CLI

pc workspace create works out which organization to use, and creates it if it doesn't exist
yet. pc workspace details prints a table instead of a debug dump.

⚠️ Deprecations

Everything below still works in 0.11.0 but raises a DeprecationWarning. If you run your test
suite with -W error, it will fail until you migrate.

Deprecated	Use instead
Workspace.setup(...)	Workspace.create(...) then .connect_provider()
Workspace.deploy(...)	Workspace.connect_provider(...)
Workspace.status	ws.aws.is_connected()
Workspace.is_active()	ws.aws.is_connected()
Workspace.wait_until_active(...)	ws.aws.wait_until_connected(...)
WorkspaceStatus	the AWS connection status on ws.aws
pc workspace verify	pc workspace aws verify
pc workspace setup	pc workspace create --connect-aws

A few smaller changes in the same area can break scripts that parse output:

• repr(Workspace) no longer includes the status field.
• pc workspace list dropped its STATUS column; the columns are now name, id and organization.
• ComputeContext.select() no longer filters workspaces by status, so it lists compute across
  all your workspaces.

Try with Polars Cloud - https://cloud.pola.rs/
On-Prem Releases - https://docs.cloud.pola.rs/polars-on-premises/releases/