Method benchmark_query

benchmarks/src/smj.rs:581–622

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

    }

    async fn benchmark_query(
        &self,
        sql: &str,
        query_name: &str,
        expect_mark: bool,
        ctx: &SessionContext,
    ) -> Result<Vec<QueryResult>> {
        let mut query_results = vec![];

        // Validate that the query plan includes a Sort Merge 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("SortMergeJoinExec") {
            return Err(exec_datafusion_err!(
                "Query {query_name} does not use Sort Merge Join. Physical plan: {plan_string}"
            ));
        }

        if expect_mark && !plan_string.contains("LeftMark") {
            return Err(exec_datafusion_err!(
                "Query {query_name} expected LeftMark 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.