andygrove opened a new issue, #5617:
URL: https://github.com/apache/datafusion-comet/issues/5617
## Describe the bug
#5370 gave the native shuffle writer a `memory_spilled_bytes` metric, which
feeds Spark's `memoryBytesSpilled` task metric through the completion listener
in `CometNativeShuffleWriter`. It dedups Arrow zero-copy slices so that one
backing allocation is charged once rather than once per slice.
That dedup only covers slices produced **within** a single `insert_batch`
call. `spill_accounted_input_buffers` is cleared at both the entry and the exit
of `insert_batch` (`native/shuffle/src/partitioners/multi_partition.rs:606` and
`:624`), so the `repeated_spill_buffer_bytes` subtraction at `:585` never
applies across calls. `pinned_buffers` cannot compensate because `spill()`
clears it at `:587`.
The uncovered case is the one the doc comment on `count_new_buffers` names
as motivating (`:141`):
> Cheaper measures do not match resident memory for the batches this writer
sees. A partial `HashAggregate` emits one group-values buffer sliced into
batch_size chunks, and every buffered chunk shares that one allocation
Those chunks reach the writer as separate `RecordBatch`es, so each is its
own `insert_batch` call. `pinned_buffers` exists precisely to dedup them for
the reservation. The spill metric does not.
## To Reproduce
Against `2949fd0d2`, driving the identical 16 chunks of one 128,168-byte
`Int64` allocation through `MultiPartitionShuffleRepartitioner` with
`max_buffer_bytes = 8 KiB` (16 spills either way), varying only how the chunks
are delivered:
| Delivery | `memory_spilled_bytes` | Ratio to the
real allocation |
| --------------------------------- | ---------------------- |
---------------------------- |
| 16 separate `insert_batch` calls | 2,228,224 | 17.0x
|
| one `insert_batch`, sliced 16 ways | 262,144 | 2.0x
|
Same allocation, same resident footprint, same spill count, 8.5x apart.
## Expected behavior
Both shapes should report the same figure, since they describe the same
resident memory. The inflated number reaches users as `memoryBytesSpilled` in
the Stages tab and as `memory spilled bytes` on the Exchange node.
## Additional context
#5370 states the cumulative-per-input-batch behavior is deliberate, and
`max_buffer_spills_charge_shared_backing_once_per_input_batch` asserts it. The
concern is that the result then depends on whether the producer emitted one
batch or sixteen, which is not a distinction Spark's spill metrics carry.
A fix is not just a matter of widening the set's lifetime. Within one
`insert_batch` the outer `batch` argument keeps the allocation alive, which is
what makes the address-keyed set sound there. Across calls nothing holds it, so
a retained address could collide with a recycled allocation. Keying on
something that pins the buffer, or scoping the set to the buffers reachable
from `buffered_batches`, would both need thought about the cost.
Related: #3996 (native shuffle observability), #5212 (memory pool and
accounting audit sweep).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
