fix: output_bytes metric in hash aggregation

[ariel-miculas]

Which issue does this PR close?

• Closes Incorrect output_bytes metric in hash aggregation #23776.

Rationale for this change

Summing together the produced sliced RecordBatches in hash aggregation leads to inflated output_bytes metric

What changes are included in this PR?

Are these changes tested?

Yes

Are there any user-facing changes?

If the metrics are user-facing, then yes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 85.78680% with 28 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.98%. Comparing base (db0c31b) to head (8995ed6).
⚠️ Report is 23 commits behind head on main.

Files with missing lines	Patch %	Lines
...fusion/physical-plan/src/aggregates/hash_stream.rs	81.20%	7 Missing and 21 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #23777      +/-   ##
==========================================
+ Coverage   80.89%   80.98%   +0.08%     
==========================================
  Files        1102     1104       +2     
  Lines      376111   378992    +2881     
  Branches   376111   378992    +2881     
==========================================
+ Hits       304251   306921    +2670     
- Misses      53753    53829      +76     
- Partials    18107    18242     +135     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Jithendra2608]

Hi @ariel-miculas! I was taking a look at the failing CI checks to see if I could help unblock this.
It looks like the typos job is failing because of a small spelling error in datafusion/physical-expr-common/src/metrics/baseline.rs on line 326 (accross instead of across).

Fixing that one word should turn the CI green. Hope this helps, and great work on the hash aggregation metrics!

[ariel-miculas]

@2010YOUY01 coud you please take a look?

[2010YOUY01]

Is it possible to implement it entirely within the record_output, and avoid large changes to different operators inner logic?

Also let's add some end-to-end tests, ideally we can do it with sqllogictest

[ariel-miculas]

Is it possible to implement it entirely within the record_output, and avoid large changes to different operators inner logic?

I don't see how, since RecordBatch::record_output doesn't keep any state related to previous RecordBatches that have been processed, and we need this information so that we can deduplicate the sliced RecordBatches to avoid summing up the memory for the same underlying original large RecordBatch.
This is a limitation that we need to deal with until we get #15591

Also let's add some end-to-end tests, ideally we can do it with sqllogictest

Added an sqllogictest

[ariel-miculas]

Closed and reopen to see if the test failure in an unrelated ordered_aggregate_spill.slt is transient:

External error: 1 errors in file /__w/datafusion/datafusion/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt

1. query failed: DataFusion error: Resources exhausted: Additional allocation failed for OrderedFinalAggregateStream[1] with top memory consumers (across reservations) as:
  OrderedFinalAggregateStream[0]#39115(can spill: true) consumed 406.8 KB, peak 496.2 KB,
  OrderedFinalAggregateStream[1]#39112(can spill: true) consumed 180.0 KB, peak 180.0 KB,
  OrderedFinalAggregateStream[1]#39235(can spill: false) consumed 5.4 KB, peak 11.5 KB.
Error: Failed to allocate additional 8.1 KB for OrderedFinalAggregateStream[1] with 5.4 KB already allocated for this reservation - 7.7 KB remain available for the total memory pool: greedy(used: 592.3 KB, pool_size: 600.0 KB)
[SQL] EXPLAIN ANALYZE
SELECT round(v1, -4), v1 % 5000,
  sum(v1 * 2), min(CAST(v1 % 2 AS VARCHAR))
FROM generate_series(20000) AS t1(v1)
GROUP BY round(v1, -4), v1 % 5000
at /__w/datafusion/datafusion/datafusion/sqllogictest/test_files/ordered_aggregate_spill.slt:166



Error: Execution("1 failures\n\nbacktrace:    0: sqllogictests::run_tests::{closure#0}\n   1: <tokio::runtime::park::CachedParkThread>::block_on::<sqllogictests::run_tests::{closure#0}>::{closure#0}\n   2: <tokio::runtime::park::CachedParkThread>::block_on::<sqllogictests::run_tests::{closure#0}>\n   3: <tokio::runtime::context::blocking::BlockingRegionGuard>::block_on::<sqllogictests::run_tests::{closure#0}>\n   4: <tokio::runtime::scheduler::multi_thread::MultiThread>::block_on::<sqllogictests::run_tests::{closure#0}>::{closure#0}\n   5: tokio::runtime::context::runtime::enter_runtime::<<tokio::runtime::scheduler::multi_thread::MultiThread>::block_on<sqllogictests::run_tests::{closure#0}>::{closure#0}, core::result::Result<(), datafusion_common::error::DataFusionError>>\n   6: <tokio::runtime::scheduler::multi_thread::MultiThread>::block_on::<sqllogictests::run_tests::{closure#0}>\n   7: <tokio::runtime::runtime::Runtime>::block_on_inner::<sqllogictests::run_tests::{closure#0}>\n   8: <tokio::runtime::runtime::Runtime>::block_on::<sqllogictests::run_tests::{closure#0}>\n   9: sqllogictests::main\n  10: <fn() -> core::result::Result<(), datafusion_common::error::DataFusionError> as core::ops::function::FnOnce<()>>::call_once\n  11: std::sys::backtrace::__rust_begin_short_backtrace::<fn() -> core::result::Result<(), datafusion_common::error::DataFusionError>, core::result::Result<(), datafusion_common::error::DataFusionError>>\n  12: std::rt::lang_start::<core::result::Result<(), datafusion_common::error::DataFusionError>>::{closure#0}\n  13: std::rt::lang_start_internal\n  14: std::rt::lang_start::<core::result::Result<(), datafusion_common::error::DataFusionError>>\n  15: main\n  16: <unknown>\n  17: __libc_start_main\n  18: _start\n")
error: test failed, to rerun pass `-p datafusion-sqllogictest --test sqllogictests`

Caused by:
  process didn't exit successfully: `/__w/datafusion/datafusion/target/llvm-cov-target/ci/deps/sqllogictests-917a8f0bbb363691` (exit status: 1)
error: process didn't exit successfully: `/usr/local/rustup/toolchains/1.97.0-x86_64-unknown-linux-gnu/bin/cargo test --profile ci --manifest-path /__w/datafusion/datafusion/Cargo.toml --target-dir /__w/datafusion/datafusion/target/llvm-cov-target --exclude datafusion-examples --exclude ffi_example_table_provider --exclude datafusion-cli --workspace --lib --tests --bins --features serde,avro,json,backtrace,integration-tests,parquet_encryption,substrait` (exit status: 1)