Upgrade Guides#

Ballista 55.0.0#

Note: Ballista 55.0.0 has not been released yet. The information provided in this section pertains to features and changes that have already been merged to the main branch and are awaiting release in this version.

Planner and execution behavior changes#

ballista.scheduler.max_partitions_per_task now defaults to 0 (unbounded)#

The default was 1, which dispatched one task per input partition. It is now 0, so the scheduler fills each task up to the assigned executor’s free vcore count: a 16-partition stage runs as one task on an idle 16-vcore executor, or as four 4-partition tasks across four 4-vcore executors.

Expect fewer, larger tasks and higher per-task memory use, since each task now holds several partitions at once — if you size the executor memory pool manually (--memory-pool-size), review it. Stages that collapse all input into a single partition are unaffected.

Set ballista.scheduler.max_partitions_per_task = 1 to restore the previous behavior.

ballista.optimizer.broadcast_join_threshold_bytes now defaults to 128 MB#

The default was 10 MB. More joins are now promoted to a broadcast (CollectLeft) join, which removes a shuffle. The build side is replicated into every concurrent probe task, so peak memory per executor is roughly this value times the number of tasks it runs at once.

Set ballista.optimizer.broadcast_join_threshold_bytes = 10485760 to restore the previous behavior, or 0 to disable broadcast promotion entirely.

Joins may shuffle their build side first to measure it#

ballista.optimizer.stage_build_side is new and defaults to true. When a join’s build side has only an estimated size and the probe side is far larger, AQE now shuffles the build side on its own, reads back its measured size, and only then chooses between a broadcast and a repartitioned join. A filtered dimension scan whose size was over-estimated can now be broadcast instead of forcing a shuffle of the fact table it joins against.

Expect an extra stage for such joins, and plans that differ from 54.x. When staging fires, the build side’s shuffle runs before the probe side’s rather than alongside it. That is bounded by ballista.optimizer.stage_build_side_min_probe_ratio (default 10) and ballista.optimizer.stage_build_side_max_estimate_multiple (default 32).

Set ballista.optimizer.stage_build_side = false to restore the previous behavior.

SortMergeJoinExec is no longer converted to a broadcast join#

54.0.0 added a static-planner rewrite that converted a SortMergeJoinExec with a small build side into a broadcast CollectLeft hash join, governed by ballista.optimizer.broadcast_sort_merge_join_enabled (default true). That rewrite produced incorrect results for some plans, so both the rewrite and its config key are removed in 55.0.0.

A SortMergeJoinExec now always executes as a sort-merge join over repartitioned inputs. Broadcast promotion still applies to a HashJoinExec whose smaller side fits under ballista.optimizer.broadcast_join_threshold_bytes (now 128 MB, see above).

This affects the default configuration. Ballista sets datafusion.optimizer.prefer_hash_join = false by default (see #1648), so joins are planned as SortMergeJoinExec unless you opt out. Under 54.0.0 those joins were then silently converted to broadcast hash joins whenever one side fit under the threshold, so expect join plan shape — and therefore performance and shuffle behavior — to change on upgrade for any query with a join whose smaller side is under 10 MB.

Alongside the incorrect results, the conversion also worked against the reason sort-merge join is the default. DataFusion’s hash join has no spill support, so a CollectLeft build side must fit in memory in every parallel task on an executor; sort-merge join spills. Converting a sort-merge join into a broadcast hash join reintroduced exactly the memory behavior that the prefer_hash_join = false default exists to avoid.

Action required: the key ballista.optimizer.broadcast_sort_merge_join_enabled no longer exists, and Ballista rejects unknown configuration keys. If you set it (including to false, which was the workaround for the incorrect results), remove it — otherwise the session fails with:

configuration key `optimizer.broadcast_sort_merge_join_enabled` does not exist

If you relied on it being false, no replacement is needed: that is now the only behavior. If you relied on it being true and want small joins to broadcast, set datafusion.optimizer.prefer_hash_join = true so joins are planned as a HashJoinExec, which remains eligible for broadcast promotion. Note this opts you into the non-spilling hash join for all joins in the session, not only the small ones.

Schedulers and executors must be upgraded together#

55.0.0 introduces BALLISTA_PROTOCOL_VERSION, a strict-equality check on the executor↔scheduler wire format. The scheduler compares its own compiled-in value against ExecutorRegistration.ballista_protocol_version on every registration, heartbeat, and poll_work, and rejects a mismatch with Status::failed_precondition:

protocol version mismatch: scheduler=3, executor=2

A rejected executor never receives work. A 54.x executor sends no version at all, which arrives as the proto default 0 and never matches a real scheduler, so a 55.0.0 scheduler rejects every 54.x executor on this same path. This prevents schedulers and executors from exchanging information they think is correct, but silently produces incorrect results.

This does not preclude rolling upgrades. A Ballista generation is internally consistent and is replaced as a unit: bump the scheduler and executor image tags in the same change and apply them together, and each rolls under its own strategy while the fleet keeps serving. See Rolling both Deployments together for the two strategies and why they differ — in particular why the scheduler wants maxUnavailable: 1 so the outgoing pod goes first, and why the client-facing Service gates on /readyz so no query is submitted to a scheduler whose executors have not yet caught up.

Note also that the version has already moved twice within the 55.0.0 development cycle (1 → 2 when the statistics sketch changed, 2 → 3 when the shuffle fetch action gained byte-range addressing). If you track main rather than releases, expect to move both halves together more than once.

The failure is loud on both ends. The scheduler logs each rejection at info and, with the prometheus feature enabled, increments ballista_scheduler_rejected_by_protocol_version_total. The executor treats a rejected heartbeat as a heartbeat failure and shuts itself down after five consecutive ones, so a stale executor is restarted by its orchestrator until its image is bumped to match rather than sitting idle in a ready state.

Clients and schedulers must have the same major version#

Each Ballista major version moves to a new DataFusion major version, and DataFusion’s plan encoding can change between majors in ways that still decode but mean something different. A client and a scheduler from different major versions could run a query and return wrong results, so from 55.0.0 they refuse to work together. This applies to the Python client too: the ballista package on PyPI must have the same major version as the cluster.

A 55.0.0 client sends its version in a ballista-version gRPC header when it submits a job, and the scheduler rejects the job with Status::failed_precondition if the major versions differ. A 54.x client sends no version, so a 55.0.0 scheduler rejects every job it submits:

Ballista client older than 55.0.0 (no `ballista-version` header) cannot be used
with scheduler 55.0.0. The client and the scheduler must have the same major
version.

The scheduler sends its own version back in the same header. A 55.0.0 client that gets no version back is talking to a 54.x scheduler, which has already queued the job without checking. The client cancels that job and returns an error. A different minor or patch version is not an error, and the client only logs a warning.

Ballista keeps clients and schedulers of the same major version compatible. Any 55.x client works with any 55.x cluster, and a change that would break an older client waits for the next major release.

If you wrote your own client against the scheduler’s gRPC API, it must send the header on every ExecuteQuery and ExecuteQueryPush request, set to the Ballista version it was built against. For a tonic request, ballista_core::version::insert_version_header does this.

API changes#

SchedulerConfig now carries a scheduler identifier#

SchedulerConfig has a new scheduler_id: String field used as the scheduler’s unique identity in scheduler state and history event logs. Its autogenerated default follows the same UUID-backed instance identity used for executor ids. It is stored separately from the scheduler callback endpoint built from external_host and bind_port. SchedulerConfig::scheduler_name() is renamed to SchedulerConfig::scheduler_endpoint(). Code that constructs SchedulerConfig with a struct literal must provide this field. Prefer the existing constructors/builders where possible; use SchedulerConfig::with_scheduler_id(...) when an embedding needs a stable scheduler id. The scheduler CLI accepts --id.

ExecutorProcessConfig similarly has a new executor_id: String field used as the executor’s unique identity when it registers with the scheduler. The executor CLI accepts --id; if unset, executor startup generates the same UUID-backed instance identity as before. Code that constructs ExecutorProcessConfig with a struct literal must provide this field or use the existing default. Active executors registered with the same scheduler must have unique ids; duplicate active ids are rejected and can be reused after the previous executor is removed from scheduler state.

The generated protobuf fields LaunchTaskParams.scheduler_id and LaunchMultiTaskParams.scheduler_id are renamed to scheduler_endpoint because they carry scheduler callback endpoints. The protobuf field number remains 2.

Shuffle fetch addresses byte ranges, not just partitions#

The shuffle fetch action carried a bool saying whether the producer used the sort-based writer. It now carries which layout the producer wrote, which of that output’s files is wanted, and optionally which bytes of it — enough for a reader that has fetched an index to ask for the bytes its own value range covers.

ballista_core::serde::scheduler::Action::FetchPartition loses is_sort_shuffle: bool and gains three fields:

  • layout: ShuffleLayout — Passthrough or Sort, how the writer laid its output out on disk. Replaces the bool: false becomes ShuffleLayout::Passthrough, true becomes ShuffleLayout::Sort.

  • file_kind: ShuffleFileKind — Data or the Index beside it.

  • byte_ranges: Vec<ByteRange> — absolute half-open ranges of that file, returned concatenated in request order. Empty asks for whatever the identifiers above address, which is the previous behavior.

All three types are new in ballista_core::serde::scheduler. On the wire, ballista.FetchPartition.is_sort_shuffle is removed and its field number reserved; layout takes a fresh number, so no old message decodes as a new one.

BallistaClient::fetch_partition and BallistaClient::fetch_partition_proxied take a ShuffleLayout where they took is_sort_shuffle: bool, in the same position. Pass ShuffleLayout::Sort where you passed true and ShuffleLayout::Passthrough where you passed false:

use ballista_core::serde::scheduler::ShuffleLayout;

client
    .fetch_partition(
        executor_id,
        &partition_id,
        file_id,
        ShuffleLayout::Passthrough, // was `false`
        flight_transport,
    )
    .await?;

BallistaClient::fetch_byte_ranges is the new entry point for a reader that knows which bytes it wants.

BallistaClient::execute_do_action is unchanged. Prefixing the returned block stream with a caller-supplied IPC schema message — which a byte-range fetch needs, since the schema does not lie within the bytes it asked for — is a separate execute_do_action_with_header(action, header).

ballista_executor::as_task_status signature reshaped#

as_task_status now takes a TaskCompletionExtras struct in place of the operator_metrics: Option<Vec<OperatorMetricsSet>> positional parameter. The struct also carries a new runtime_stats: Vec<RuntimeStatsReport> field used to transport RuntimeStatsExec reports back to the scheduler. It is marked #[non_exhaustive] with a Default impl, so future additions to the struct are non-breaking for callers that construct via ..Default::default().

The parameter order also changed: execution_times now precedes the extras struct.

Action required: external callers building a TaskStatus via as_task_status should migrate from:

as_task_status(
    execution_result,
    executor_id,
    stage_attempt_num,
    key,
    operator_metrics,
    execution_times,
)

to:

as_task_status(
    execution_result,
    executor_id,
    stage_attempt_num,
    key,
    execution_times,
    TaskCompletionExtras {
        operator_metrics,
        ..Default::default()
    },
)

SessionConfigExt and BallistaConfig Changes#

Methods

SessionConfigExt::ballista_coalesce_target_partition_bytes(&self) -> u64;
SessionConfigExt::with_ballista_coalesce_target_partition_bytes(self, bytes: u64) -> Self;

Changed expected type from u64 to usize

SessionConfigExt::ballista_coalesce_target_partition_bytes(&self) -> usize;
SessionConfigExt::with_ballista_coalesce_target_partition_bytes(self, bytes: usize) -> Self;

the change is to align method parameter and return type to actual config value.

Also,

BallistaConfig::coalesce_target_partition_bytes(&self) -> u64;

has return type changed from u64 to usize

BallistaConfig::coalesce_target_partition_bytes(&self) -> usize;