Method run

benchmarks/src/smj.rs:528–579

(self)

impl RunOpt {
    pub async fn run(self) -> Result<()> {
        println!("Running SMJ benchmarks with the following options: {self:#?}\n");

        // Define query range
        let query_range = match self.query {
            Some(query_id) => {
                if query_id >= 1 && query_id <= SMJ_QUERIES.len() {
                    query_id..=query_id
                } else {
                    return exec_err!(
                        "Query {query_id} not found. Available queries: 1 to {}",
                        SMJ_QUERIES.len()
                    );
                }
            }
            None => 1..=SMJ_QUERIES.len(),
        };

        let mut config = self.common.config()?;
        // Disable hash joins to force SMJ
        config = config.set_bool("datafusion.optimizer.prefer_hash_join", false);
        let rt = self.common.build_runtime()?;
        let ctx = SessionContext::new_with_config_rt(config, rt);

        let mut benchmark_run = BenchmarkRun::new();
        for query_id in query_range {
            let query_index = query_id - 1; // Convert 1-based to 0-based index

            let sql = SMJ_QUERIES[query_index];
            benchmark_run.start_new_case(&format!("Query {query_id}"));
            let expect_mark = query_id >= 24;
            let query_run = self
                .benchmark_query(sql, &query_id.to_string(), expect_mark, &ctx)
                .await;
            match query_run {
                Ok(query_results) => {
                    for iter in query_results {
                        benchmark_run.write_iter(iter.elapsed, iter.row_count);
                    }
                }
                Err(e) => {
                    return Err(DataFusionError::Context(
                        format!("SMJ benchmark Q{query_id} failed with error:"),
                        Box::new(e),
                    ));
                }
            }
        }

        benchmark_run.maybe_write_json(self.output_path.as_ref())?;
        Ok(())
    }

    async fn benchmark_query(
        &self,