perf(core): zero-copy block receive on the remote fetch path

[Dandandan]

Which issue does this PR close?

N/A

Rationale for this change

BlockDataStream copied every incoming transport block into a freshly allocated buffer, even when nothing was pending:

self.state_buffer = Self::combine_buffers(&self.state_buffer, &Buffer::from(blob))

Two facts make that copy avoidable in the steady state:

• Buffer::from(bytes::Bytes) adopts the transport allocation rather than copying it.
• StreamDecoder::decode loops while !buffer.is_empty(), and every non-error early return is Ok(Some(batch)). So whenever it returns Ok(None) — the only condition under which poll_next pulls another block — state_buffer has been fully drained.

So on the hot path the incoming block can simply be adopted. Concatenation is only needed for a partial message straddling a block boundary, which is what the schema-accumulation loop in try_new relies on (try_schema_from_ipc_buffer peeks without consuming, so a partial header stays pending there).

What changes are included in this PR?

combine_buffers is replaced by append_block, which adopts the incoming block when nothing is pending and otherwise falls back to the existing concatenating path. Both call sites — the schema loop in try_new and extend_bytes — move to it.

This is on by default for all remote shuffle reads; there is no new configuration.

Are these changes tested?

Yes. The existing BlockDataStream tests cover both paths and still pass — should_process_chunked (2-byte blocks) drives the concatenating path in the schema loop, and should_process_single_message covers whole-block adoption.

Added should_process_multi_block_payload, which round-trips a payload spanning several whole blocks at 8/64/512-byte block sizes, covering the mixed regime this change targets.

cargo test --package ballista-core --lib client:: — 6 passed, 0 failed. cargo clippy --all-targets --package ballista-core --all-features -- -D warnings and cargo fmt --all -- --check are clean.

Benchmark

65 MiB payload at the server's 8 MiB block size: 69.7 ms → 22.7 ms (−67%). At 1 MiB blocks: 41.8 ms → 34.9 ms (−17%).

Are there any user-facing changes?

No API changes — combine_buffers and append_block are both private. Remote shuffle reads get faster.

🤖 Generated with Claude Code

[milenkovicm]

Thanks @Dandandan makes sense