perf: grow the native shuffle writer's reservation in steps

[andygrove]

Which issue does this PR close?

Part of #5383, following the measurements in #5383 (comment). Independent of #5384 (different crates, either merge order works).

Rationale for this change

buffer_partitioned_batch_may_spill reserved exactly what each batch newly pinned. For an 8192-row i32 column that is 32 KB, and against Comet's unified memory pool every try_grow is a JNI round-trip into Spark's memory manager plus contention on its executor-wide lock.

Measuring a 201-partition shuffle of 10M rows with the counters from #5384, ShuffleRepartitioner[0] was the only consumer touching the pool, and it made 256 acquisitions per task to reserve 24 MB — 247 of them for exactly 32 KB — followed by a single release of the whole amount. Nothing about that pattern needs a round-trip per batch.

What changes are included in this PR?

The repartitioner tracks the exact resident figure itself in buffered_bytes and grows reservation in steps on top of it, so a run of small batches costs one acquisition instead of one each. The step is spark.comet.shuffle.native.reservationStepBytes, defaulting to 1 MB; zero grows by exactly what each batch needs, which is the pre-change behaviour.

Two details keep the spill behaviour where it was:

• max_buffer_bytes is compared against buffered_bytes rather than reservation.size(). The reservation is now rounded up, so comparing against it would trip the limit early and spill sooner than before.
• if the pool denies a rounded-up request, it is retried at the exact deficit before spilling. A batch that fit before still fits; the extra call happens only on the path that was about to spill anyway.

used() reports buffered_bytes, which is the same number it reported before this change (the exact resident bytes), so the spill log is unaffected.

The two memory knobs now travel together in a ShuffleWriterMemoryConfig instead of as more positional arguments — ShuffleWriterExec::try_new and external_shuffle already carry a write_buffer_size that is easy to transpose with a max_buffer_bytes or a step, and all three are byte counts.

The cost of a non-zero step is up to that much reservation held beyond what a writer is using, per concurrent shuffle-write task. The default of 1 MB keeps that bounded while still removing 95% of the calls, and the step is capped at max_buffer_bytes / 8 so a writer with a small limit cannot reserve a multiple of its own budget.

How are these changes tested?

Two new unit tests, over a MemoryPool that counts the try_grow calls reaching it:

• small_batches_share_one_reservation_step — 40 distinct batches of 100 rows produce 1 acquisition at the default step
• zero_reservation_step_reserves_per_batch — the same input produces exactly 40 with the step set to zero

The pair pins both that the reduction comes from the step and that the config is honoured end to end. The batches are built separately rather than cloned, because count_new_buffers dedups by buffer address and re-inserting one batch would charge nothing after the first insert.

Existing coverage: cargo test -p datafusion-comet-shuffle --lib (33 tests) passes, including max_buffer_bytes_triggers_spill_without_memory_pressure, max_buffer_bytes_none_leaves_spilling_to_memory_pressure and max_buffer_bytes_preserves_output, which guard the limit semantics this touches. Also green: cargo test -p datafusion-comet --lib (163), CometNativeShuffleSuite (27), CometConfSuite (13), cargo clippy --workspace --all-targets, cargo fmt --check, spotless:check.

Measured effect

Same benchmark case as the issue comment (SQL Single INT Shuffle(201 Partition), 10M rows, native shuffle, release build, 4 GB off-heap, default fair_unified, local[5]), with the #5384 counters enabled and the default 1 MB step:

Per task	Before	After
Acquire calls	256	13
Release calls	1	1
Time inside the JNI call	327 µs	50 µs
Bytes reserved	25,165,824	25,198,592

End-to-end wall time is not a useful signal here and I do not want to claim one: the saving is ~0.3 ms out of a ~91 ms task, and between the two runs every case moved by 5-10% including the pure-Spark baseline that this change cannot affect (620 -> 659 ms). The case for the change rests on the call count and the JNI time, both measured directly, plus the reduced traffic on Spark's shared memory-manager lock at higher concurrency than a local benchmark produces.