Add `output_bytes` metrics to Explain Analyze

[PokIsemaine]

Is your feature request related to a problem or challenge?

Currently, using Explain Analyze seems to only provide the metric for output_rows. Is it possible to add a metric for the number of bytes output by each operator?

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[2010YOUY01]

It's a good idea.

A quick reminder for someone who is willing to implement it: It's possible that multiple Arrays share the same underlying buffer -- those Arrays can be within the same batch or across batches.
The current memory accounting utilities only support detect buffer reuse within the batch (not across batches), so a extension to this might be needed.

datafusion/datafusion/physical-plan/src/spill/mod.rs

Line 239 in d68111d

pub fn get_record_batch_memory_size(batch: &RecordBatch) -> usize {

[hendrikmakait]

I'd be interested in working on this, but I might need a little guidance since I'm new to the project.

[2010YOUY01]

I'd be interested in working on this, but I might need a little guidance since I'm new to the project.

Thank you! Here are some additional info

Each operator holds a BaselineMetrics inside for common metrics like output_rows, and this new output_bytes should also belong to BaselineMetrics.
When the operator output one batch, it will call record_poll() to update the BaselineMetrics

datafusion/datafusion/physical-plan/src/metrics/baseline.rs

Line 123 in 992d156

pub fn record_poll(

So I think the implementation would be adding a new field output_bytes (and other useful structures) into BaselineMetrics, and then update them inside record_poll(), the tricky part would be avoiding double-counting array buffers I mentioned above.

To see the expected result, run a explain analyze query using datafusion-cli, and this new metrics should show up inside metrics

> explain analyze select * from generate_series(1, 1000000) as t1(v1) order by v1 desc;
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type         | plan                                                                                                                                                                      |
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | SortExec: expr=[v1@0 DESC], preserve_partitioning=[false], metrics=[output_rows=1000000, elapsed_compute=91.856373ms, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0] |
|                   |   ProjectionExec: expr=[value@0 as v1], metrics=[output_rows=1000000, elapsed_compute=14.702µs]                                                                           |
|                   |     LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=1, end=1000000, batch_size=8192], metrics=[]                                                   |
|                   |                                                                                                                                                                           |
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.039 seconds.

[hendrikmakait]

FYI, unfortunately I haven't been had the time to implement this yet and won't be around next week. I plan to pick this up afterward unless someone beats me to it.

[hendrikmakait]

take

[hendrikmakait]

A quick reminder for someone who is willing to implement it: It's possible that multiple Arrays share the same underlying buffer -- those Arrays can be within the same batch or across batches. The current memory accounting utilities only support detect buffer reuse within the batch (not across batches), so a extension to this might be needed.

datafusion/datafusion/physical-plan/src/spill/mod.rs

Line 239 in d68111d

pub fn get_record_batch_memory_size(batch: &RecordBatch) -> usize {

I'm wondering about a good approach here. get_record_batch_memory_size uses the data pointers of the buffers to deduplicate. Within a single record batch, that doesn't seem problematic. I'm not sure if we can use the same approach across multiple record batches. If I understand correctly, those record batches do not have the same lifetime, so the same pointer could be allocated twice and filled with different data. (Can Buffer objects themselves be reused with different data?) In this case, we want to count the pointer/buffer twice. That being said, this scenario might happen infrequently enough for us to not care.

There's also some discussion around this going on over at apache/arrow-rs#6439 which makes it sound like accurately measuring bytes in this scenario is a bigger issue. @2010YOUY01, do you have thoughts on this?

An alternative approach could be to calculate output_bytes based on the actual number of bytes occupied on the buffer by the array. This should be accurate but might be expensive.

[2010YOUY01]

A quick reminder for someone who is willing to implement it: It's possible that multiple Arrays share the same underlying buffer -- those Arrays can be within the same batch or across batches. The current memory accounting utilities only support detect buffer reuse within the batch (not across batches), so a extension to this might be needed.
datafusion/datafusion/physical-plan/src/spill/mod.rs
Line 239 in d68111d
pub fn get_record_batch_memory_size(batch: &RecordBatch) -> usize {

I'm wondering about a good approach here. get_record_batch_memory_size uses the data pointers of the buffers to deduplicate. Within a single record batch, that doesn't seem problematic. I'm not sure if we can use the same approach across multiple record batches. If I understand correctly, those record batches do not have the same lifetime, so the same pointer could be allocated twice and filled with different data. (Can Buffer objects themselves be reused with different data?) In this case, we want to count the pointer/buffer twice. That being said, this scenario might happen infrequently enough for us to not care.

There's also some discussion around this going on over at apache/arrow-rs#6439 which makes it sound like accurately measuring bytes in this scenario is a bigger issue. @2010YOUY01, do you have thoughts on this?

An alternative approach could be to calculate output_bytes based on the actual number of bytes occupied on the buffer by the array. This should be accurate but might be expensive.

I agree with the concerns about the buffer lifetime across batches. Another problem might be: if the input is very large this hashset-based solution might cause noticeable slowdown.

I think we need an arrow-side solution that better solve this issue. This refcount solution seems a good idea to me, but it requires more effort to move it forward apache/arrow-rs#6590

Now for datafusion, maybe the best thing to do is simply sum(get_record_batch_memory_size()), this should work for common cases, and open an github issue to explain the limitation and track future TODOs to make it accurate.

[hendrikmakait]

Now for datafusion, maybe the best thing to do is simply sum(get_record_batch_memory_size()), this should work for common cases, and open an github issue to explain the limitation and track future TODOs to make it accurate.

Sounds like a good plan to me, I'll get started.

[hendrikmakait]

It looks like there are other sets of metrics that would benefit from adding output_bytes for improved data in EXPLAIN ANALYZE, namely UnnestMetrics, SortMergeJoinMetrics, StreamJoinMetrics, and BuildProbeJoinMetrics. I'm wondering whether we should add output_bytes to these structs or rather refactor them to make use of BaselineMetrics.

[2010YOUY01]

It looks like there are other sets of metrics that would benefit from adding output_bytes for improved data in EXPLAIN ANALYZE, namely UnnestMetrics, SortMergeJoinMetrics, StreamJoinMetrics, and BuildProbeJoinMetrics. I'm wondering whether we should add output_bytes to these structs or rather refactor them to make use of BaselineMetrics.

If they're the top-level metrics struct directly managed by an executor (xxExec), then it should be better to let then use BaselineMetrics.
I'll double-check and file tickets, thanks for the investigation.