Validates that the physical plan uses a HashJoin, then executes.
(
&self,
sql: &str,
query_name: &str,
ctx: &SessionContext,
)
| 386 | |
| 387 | /// Validates that the physical plan uses a HashJoin, then executes. |
| 388 | async fn benchmark_query( |
| 389 | &self, |
| 390 | sql: &str, |
| 391 | query_name: &str, |
| 392 | ctx: &SessionContext, |
| 393 | ) -> Result<Vec<QueryResult>> { |
| 394 | let mut query_results = vec![]; |
| 395 | |
| 396 | // Build/validate plan |
| 397 | let df = ctx.sql(sql).await?; |
| 398 | let physical_plan = df.create_physical_plan().await?; |
| 399 | let plan_string = format!("{physical_plan:#?}"); |
| 400 | |
| 401 | if !plan_string.contains("HashJoinExec") { |
| 402 | return Err(exec_datafusion_err!( |
| 403 | "Query {query_name} does not use Hash Join. Physical plan: {plan_string}" |
| 404 | )); |
| 405 | } |
| 406 | |
| 407 | // Execute without buffering |
| 408 | for i in 0..self.common.iterations { |
| 409 | let start = Instant::now(); |
| 410 | let row_count = Self::execute_sql_without_result_buffering(sql, ctx).await?; |
| 411 | let elapsed = start.elapsed(); |
| 412 | |
| 413 | println!( |
| 414 | "Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}" |
| 415 | ); |
| 416 | |
| 417 | query_results.push(QueryResult { elapsed, row_count }); |
| 418 | } |
| 419 | |
| 420 | Ok(query_results) |
| 421 | } |
| 422 | |
| 423 | /// Executes the SQL query and drops each batch to avoid result buffering. |
| 424 | async fn execute_sql_without_result_buffering( |
no test coverage detected