fix: propagate sort shuffle memory_limit_per_task_bytes to executors#2090
Conversation
The sort shuffle writer's per-task buffered-bytes budget (`ballista.shuffle.sort_based.memory_limit_per_task_bytes`) was built into `SortShuffleConfig` by the scheduler but never carried in the `SortShuffleWriterExecNode` protobuf message. Executors, where the writer actually runs, reconstructed the config with the built-in default, so the session override was inert in distributed execution. Carry the value on the wire as an optional field (absent decodes to the built-in default) and apply it on decode. Also treat a value of 0 as "disable the per-task budget", leaving runtime memory-pool pressure as the sole spill trigger, and refresh the tuning guide and configs docs, which still referenced removed keys. Closes apache#2089
metegenez
left a comment
There was a problem hiding this comment.
Only nit: adding an OOM warning to the config docs for the 0 + unbounded-pool combo. Other than that, LGTM. Test coverage looks good.
One question about having 2 spill triggers: In a world we mostly cover and account the bytes into memory pool, this local operator count and trigger wouldnt be necessary, right?
Added. Thanks.
Yes, if we had accurate memory accounting (perhaps backed by the real memory allocator - see #2033) I think we would revisit these configs. |
Which issue does this PR close?
Closes #2089.
Rationale for this change
ballista.shuffle.sort_based.memory_limit_per_task_bytes(
SortShuffleConfig::memory_limit_per_task_bytes) controls the point at whichthe sort shuffle writer spills its in-memory batches to disk. The scheduler
built
SortShuffleWriterExecwith the configured value, but the value was notcarried in the
SortShuffleWriterExecNodeprotobuf message. On decode, theexecutor reconstructed the config with
SortShuffleConfig::new(true, batch_size),which discards the override and uses the built-in default. Because the writer
actually runs on the executor, the knob was effectively inert in distributed
execution.
What changes are included in this PR?
memory_limit_per_task_bytesto theSortShuffleWriterExecNodemessageas an
optional uint64and populate it on encode. Presence lets a planproduced before this field existed decode to the built-in default rather than
to
0.with_memory_limit_per_task_bytes, so thesession override reaches the executor.
0as "disable the per-task budget": the writer then spillsonly under runtime memory-pool pressure. Previously a
0limit would haveforced a spill on every batch.
(
buffer_size,memory_limit,spill_threshold), and describe the per-taskbudget and the
0-disables behavior.Are there any user-facing changes?
The configuration setting now takes effect on executors as documented. Setting
it to
0is newly meaningful: it disables the per-task budget and relies onmemory-pool pressure to trigger spilling. No public Rust API changes.