Method benchmark_query

benchmarks/src/nlj.rs:244–278

Validates that the query's physical plan uses a NestedLoopJoin (NLJ), then executes the query and collects execution times. TODO: ensure the optimizer won't change the join order (it's not at v48.0.0).

(
        &self,
        sql: &str,
        query_name: &str,
        ctx: &SessionContext,
    )

    /// TODO: ensure the optimizer won't change the join order (it's not at
    /// v48.0.0).
    async fn benchmark_query(
        &self,
        sql: &str,
        query_name: &str,
        ctx: &SessionContext,
    ) -> Result<Vec<QueryResult>> {
        let mut query_results = vec![];

        // Validate that the query plan includes a Nested Loop Join
        let df = ctx.sql(sql).await?;
        let physical_plan = df.create_physical_plan().await?;
        let plan_string = format!("{physical_plan:#?}");

        if !plan_string.contains("NestedLoopJoinExec") {
            return Err(exec_datafusion_err!(
                "Query {query_name} does not use Nested Loop Join. Physical plan: {plan_string}"
            ));
        }

        for i in 0..self.common.iterations {
            let start = Instant::now();

            let row_count = Self::execute_sql_without_result_buffering(sql, ctx).await?;

            let elapsed = start.elapsed();

            println!(
                "Query {query_name} iteration {i} returned {row_count} rows in {elapsed:?}"
            );

            query_results.push(QueryResult { elapsed, row_count });
        }

        Ok(query_results)
    }

    /// Executes the SQL query and drops each result batch after evaluation, to
    /// minimizes memory usage by not buffering results.