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.

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.

API changes#

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;