Skip to content

Measure pool_peak_bytes per iteration and report the smallest #24023

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adriangb wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Measure pool_peak_bytes per iteration and report the smallest #24023

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions benchmarks/src/bin/benchmark_runner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,9 @@ use datafusion_benchmarks::sql_benchmark_runner::{
load_benchmark_definitions, make_ctx, prepare_benchmark,
run_criterion_benchmarks_impl, sort_benchmarks,
};
use datafusion_benchmarks::util::{BenchmarkRun, CommonOpt, print_memory_stats};
use datafusion_benchmarks::util::{
BenchmarkRun, CommonOpt, print_memory_stats, take_iteration_peak,
};
use datafusion_common::instant::Instant;
use datafusion_common::{DataFusionError, exec_datafusion_err};
use datafusion_common_runtime::SpawnedTask;
Expand Down Expand Up @@ -312,7 +314,8 @@ async fn run_simple_benchmark(

println!("{case_name} iteration {iteration}: {ms:.1} ms, {row_count} rows");

run.write_iter(elapsed, row_count);
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
run.write_iter(elapsed, row_count, pool_peak_bytes);
}

print_memory_stats(&*ctx.runtime_env().memory_pool);
Expand Down
15 changes: 12 additions & 3 deletions benchmarks/src/bin/external_aggr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ use datafusion::physical_plan::display::DisplayableExecutionPlan;
use datafusion::physical_plan::{collect, displayable};
use datafusion::prelude::*;
use datafusion_benchmarks::util::{
BenchmarkRun, CommonOpt, PeakRecordingPool, QueryResult,
BenchmarkRun, CommonOpt, PeakRecordingPool, QueryResult, take_iteration_peak,
};
use datafusion_common::instant::Instant;
use datafusion_common::utils::get_available_parallelism;
Expand Down Expand Up @@ -174,7 +174,11 @@ impl ExternalAggrConfig {
.benchmark_query(query_id, mem_limit, mem_pool_type, &mut benchmark_run)
.await?;
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}

Expand Down Expand Up @@ -242,7 +246,12 @@ impl ExternalAggrConfig {
println!(
"{query_name} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
});
}

let avg = millis.iter().sum::<f64>() / millis.len() as f64;
Expand Down
3 changes: 2 additions & 1 deletion benchmarks/src/cancellation.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,8 @@ impl RunOpt {
let elapsed = run_test(self.wait_time, Arc::clone(&store))?;
let ms = elapsed.as_secs_f64() * 1000.0;
println!("Iteration {i} cancelled in {ms} ms");
rundata.write_iter(elapsed, 0);
// No memory limit here, so no pool to read.
rundata.write_iter(elapsed, 0, None);
}

rundata.maybe_write_json(self.output_path.as_ref())?;
Expand Down
17 changes: 14 additions & 3 deletions benchmarks/src/clickbench.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,9 @@ use std::fs;
use std::io::ErrorKind;
use std::path::{Path, PathBuf};

use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
use crate::util::{
BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
};
use clap::Args;
use datafusion::logical_expr::{ExplainFormat, ExplainOption};
use datafusion::{
Expand Down Expand Up @@ -230,7 +232,11 @@ impl RunOpt {
match query_run {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -264,7 +270,12 @@ impl RunOpt {
println!(
"Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
query_results.push(QueryResult { elapsed, row_count })
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
})
}
if self.common.debug {
ctx.sql(sql)
Expand Down
15 changes: 12 additions & 3 deletions benchmarks/src/dict.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

use crate::util::BenchmarkRun;
use crate::util::CommonOpt;
use crate::util::QueryResult;
use crate::util::{QueryResult, take_iteration_peak};
use arrow::array::{ArrayRef, DictionaryArray, Int32Array, ListArray, StringArray};
use arrow::buffer::OffsetBuffer;
use arrow::datatypes::{DataType, Field, Int32Type, Schema};
Expand Down Expand Up @@ -343,7 +343,11 @@ impl RunOpt {
match query_run {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -378,7 +382,12 @@ impl RunOpt {
"Query '{}' iteration {i} returned {row_count} rows in {elapsed:?}",
query.name
);
query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
});
}

Ok(query_results)
Expand Down
5 changes: 3 additions & 2 deletions benchmarks/src/h2o.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
//! - [H2O AI Benchmark](https://duckdb.org/2023/04/14/h2oai.html)
//! - [Extended window function benchmark](https://duckdb.org/2024/06/26/benchmarks-over-time.html#window-functions-benchmark)

use crate::util::{BenchmarkRun, CommonOpt, print_memory_stats};
use crate::util::{BenchmarkRun, CommonOpt, print_memory_stats, take_iteration_peak};
use clap::Args;
use datafusion::logical_expr::{ExplainFormat, ExplainOption};
use datafusion::{error::Result, prelude::SessionContext};
Expand Down Expand Up @@ -126,7 +126,8 @@ impl RunOpt {
println!(
"Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
benchmark_run.write_iter(elapsed, row_count);
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
benchmark_run.write_iter(elapsed, row_count, pool_peak_bytes);
}
let avg = millis.iter().sum::<f64>() / millis.len() as f64;
println!("Query {query_id} avg time: {avg:.2} ms");
Expand Down
18 changes: 15 additions & 3 deletions benchmarks/src/hj.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.

use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
use clap::Args;
use datafusion::physical_plan::execute_stream;
use datafusion::{error::Result, prelude::SessionContext};
Expand Down Expand Up @@ -600,7 +600,11 @@ impl RunOpt {
match query_run {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -646,7 +650,15 @@ impl RunOpt {
"Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
);

query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);

query_results.push(QueryResult {
elapsed,

row_count,

pool_peak_bytes,
});
}

Ok(query_results)
Expand Down
17 changes: 14 additions & 3 deletions benchmarks/src/imdb/run.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,9 @@ use super::{
IMDB_QUERY_END_ID, IMDB_QUERY_START_ID, IMDB_TABLES, get_imdb_table_schema,
get_query_sql,
};
use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
use crate::util::{
BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
};

use arrow::record_batch::RecordBatch;
use arrow::util::pretty::{self, pretty_format_batches};
Expand Down Expand Up @@ -297,7 +299,11 @@ impl RunOpt {
benchmark_run.start_new_case(&format!("Query {query_id}"));
let query_run = self.benchmark_query(query_id, &mut benchmark_run).await?;
for iter in query_run {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
benchmark_run.maybe_write_json(self.output_path.as_ref())?;
Expand Down Expand Up @@ -348,7 +354,12 @@ impl RunOpt {
println!(
"Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
});
}

let avg = millis.iter().sum::<f64>() / millis.len() as f64;
Expand Down
18 changes: 15 additions & 3 deletions benchmarks/src/nlj.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.

use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
use clap::Args;
use datafusion::physical_plan::execute_stream;
use datafusion::{error::Result, prelude::SessionContext};
Expand Down Expand Up @@ -221,7 +221,11 @@ impl RunOpt {
match query_run {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -272,7 +276,15 @@ impl RunOpt {
"Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
);

query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);

query_results.push(QueryResult {
elapsed,

row_count,

pool_peak_bytes,
});
}

Ok(query_results)
Expand Down
18 changes: 15 additions & 3 deletions benchmarks/src/smj.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.

use crate::util::{BenchmarkRun, CommonOpt, QueryResult};
use crate::util::{BenchmarkRun, CommonOpt, QueryResult, take_iteration_peak};
use clap::Args;
use datafusion::physical_plan::execute_stream;
use datafusion::{error::Result, prelude::SessionContext};
Expand Down Expand Up @@ -563,7 +563,11 @@ impl RunOpt {
match query_run {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -616,7 +620,15 @@ impl RunOpt {
"Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
);

query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);

query_results.push(QueryResult {
elapsed,

row_count,

pool_peak_bytes,
});
}

Ok(query_results)
Expand Down
17 changes: 14 additions & 3 deletions benchmarks/src/sort_pushdown.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,9 @@ use datafusion::prelude::*;
use datafusion_common::DEFAULT_PARQUET_EXTENSION;
use datafusion_common::instant::Instant;

use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
use crate::util::{
BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
};

/// Default path to query files, relative to the benchmark root
const SORT_PUSHDOWN_QUERY_DIR: &str = "queries/sort_pushdown";
Expand Down Expand Up @@ -141,7 +143,11 @@ impl RunOpt {
match query_results {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -192,7 +198,12 @@ impl RunOpt {
println!(
"Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
});
}

let avg = millis.iter().sum::<f64>() / millis.len() as f64;
Expand Down
17 changes: 14 additions & 3 deletions benchmarks/src/sort_tpch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,9 @@ use datafusion_common::DEFAULT_PARQUET_EXTENSION;
use datafusion_common::instant::Instant;
use datafusion_common::utils::get_available_parallelism;

use crate::util::{BenchmarkRun, CommonOpt, QueryResult, print_memory_stats};
use crate::util::{
BenchmarkRun, CommonOpt, QueryResult, print_memory_stats, take_iteration_peak,
};

#[derive(Debug, Args)]
pub struct RunOpt {
Expand Down Expand Up @@ -191,7 +193,11 @@ impl RunOpt {
match query_results {
Ok(query_results) => {
for iter in query_results {
benchmark_run.write_iter(iter.elapsed, iter.row_count);
benchmark_run.write_iter(
iter.elapsed,
iter.row_count,
iter.pool_peak_bytes,
);
}
}
Err(e) => {
Expand Down Expand Up @@ -251,7 +257,12 @@ impl RunOpt {
println!(
"Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
);
query_results.push(QueryResult { elapsed, row_count });
let pool_peak_bytes = take_iteration_peak(&ctx.runtime_env().memory_pool);
query_results.push(QueryResult {
elapsed,
row_count,
pool_peak_bytes,
});
}

let avg = millis.iter().sum::<f64>() / millis.len() as f64;
Expand Down
Loading
Loading