Method run

benchmarks/src/nlj.rs:191–237

(self)

impl RunOpt {
    pub async fn run(self) -> Result<()> {
        println!("Running NLJ 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 <= NLJ_QUERIES.len() {
                    query_id..=query_id
                } else {
                    return exec_err!(
                        "Query {query_id} not found. Available queries: 1 to {}",
                        NLJ_QUERIES.len()
                    );
                }
            }
            None => 1..=NLJ_QUERIES.len(),
        };

        let config = self.common.config()?;
        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 = NLJ_QUERIES[query_index];
            benchmark_run.start_new_case(&format!("Query {query_id}"));
            let query_run = self.benchmark_query(sql, &query_id.to_string(), &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(
                        "NLJ benchmark Q{query_id} failed with error:".to_string(),
                        Box::new(e),
                    ));
                }
            }
        }

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

    /// Validates that the query's physical plan uses a NestedLoopJoin (NLJ),
    /// then executes the query and collects execution times.