avantgardnerio commented on PR #26861:
URL: https://github.com/apache/datafusion/pull/26861#issuecomment-6478030719
### Follow-up idea: the same accounts, one per operator
Not part of this PR, but it shows where this goes. Accounts nest, so
wrapping each operator's stream in its own account charges every operator for
its own polls and not its children's. The file and statement accounts still
balance. As a prototype, the SLT runner wraps every node of the physical plan
just before execution (about 200 lines, SLT-only).
`ExecutionPlan::downcast_delegate` already exists for this kind of
instrumentation wrapper, so downcast-based code sees the operator underneath.
A free is charged to whichever operator's poll ran it. So an operator's
number is "bytes allocated in my polls that are still live", which points at
who allocated untracked memory, not who holds it. The batches a consumer frees
show up as a matching negative on the consumer.
Another property worth noting: an operator that isn't wrapped has its
allocations charged to the nearest wrapped ancestor, and ultimately the
statement. The total stays complete, so a missing wrapper only coarsens the
attribution.
### What it found on its first run: `RepartitionExec` allocates 10.1 MB to
move 6 rows
`aggregate.slt`, grouped `regr_*` over a 6-row `VALUES`:
```sql
select column3, regr_slope(column2, column1), regr_intercept(column2,
column1), ...
from (values (1,2,'a'), (2,4,'a'), (1,3,'b'), (3,9,'b'), (1,10,'c'),
(NULL,100,'c'))
group by column3;
```
```
AggregateExec peak=200.7 KB net=-10.1 MB
RepartitionExec peak=10.1 MB net=10.1 MB
AggregateExec peak=42.9 KB net=38.4 KB
ProjectionExec peak=2.9 KB net=2.9 KB
DataSourceExec peak=344 B net=168 B
```
Each output partition of `RepartitionExec` has a `LimitedBatchCoalescer`
([`repartition/mod.rs`](https://github.com/apache/datafusion/blob/56eca4a7c5f8890c82902518c3dca4c08ae9bfb6/datafusion/physical-plan/src/repartition/mod.rs#L257-L278)).
It wraps arrow's `BatchCoalescer`, which on the first write reserves a full
`batch_size` for every primitive column, however many rows arrive
([`arrow-select/src/coalesce/primitive.rs`](https://github.com/apache/arrow-rs/blob/60.0.0/arrow-select/src/coalesce/primitive.rs#L57-L61)):
8192 × 8 bytes = 64 KB per column.
The partial `regr_*` state has 6 columns per aggregate
([`regr.rs`](https://github.com/apache/datafusion/blob/56eca4a7c5f8890c82902518c3dca4c08ae9bfb6/datafusion/functions-aggregate/src/regr.rs#L471-L497)),
so 54 columns. The 3 groups hash to 3 of the 4 output partitions: 3 × 54 × 64
KB = 10.125 MiB, which is the 10.1 MB measured. None of it is reserved in the
pool, and the final aggregate frees it.
10 MB is small here. The cost is output partitions × primitive columns ×
`batch_size` × value width for every partition that receives a row, independent
of the row count. With this schema and 64 target partitions, that is about 216
MB per repartition, regardless of how few rows are moved.
The statement nets to zero, so the per-file drift in this PR only catches it
when a sample lands at the peak. It came and went between full-suite runs, as
10.4 MB "added" by this statement in one run and nothing in the next.
Per-operator peaks catch it every time.
--
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]
