Method run

benchmarks/src/h2o.rs:78–149

(self)

impl RunOpt {
    pub async fn run(self) -> Result<()> {
        println!("Running benchmarks with the following options: {self:?}");
        let queries = AllQueries::try_new(&self.queries_path)?;
        let query_range = match self.query {
            Some(query_id) => query_id..=query_id,
            None => queries.min_query_id()..=queries.max_query_id(),
        };

        let config = self.common.config()?;
        let rt = self.common.build_runtime()?;
        let ctx = SessionContext::new_with_config_rt(config, rt);

        // Register tables depending on which h2o benchmark is being run
        // (groupby/join/window)
        if self.queries_path.to_str().unwrap().ends_with("groupby.sql") {
            self.register_data("x", self.path.as_os_str().to_str().unwrap(), &ctx)
                .await?;
        } else if self.queries_path.to_str().unwrap().ends_with("join.sql") {
            let join_paths: Vec<&str> = self.join_paths.split(',').collect();
            let table_name: Vec<&str> = vec!["x", "small", "medium", "large"];
            for (i, path) in join_paths.iter().enumerate() {
                self.register_data(table_name[i], path, &ctx).await?;
            }
        } else if self.queries_path.to_str().unwrap().ends_with("window.sql") {
            // Only register the 'large' table in h2o-join dataset
            let h2o_join_large_path = self.join_paths.split(',').nth(3).unwrap();
            self.register_data("large", h2o_join_large_path, &ctx)
                .await?;
        } else {
            return internal_err!("Invalid query file path");
        }

        let iterations = self.common.iterations;
        let mut benchmark_run = BenchmarkRun::new();
        for query_id in query_range {
            benchmark_run.start_new_case(&format!("Query {query_id}"));
            let sql = queries.get_query(query_id)?;
            println!("Q{query_id}: {sql}");

            let mut millis = Vec::with_capacity(iterations);
            for i in 1..=iterations {
                let start = Instant::now();
                let results = ctx.sql(sql).await?.collect().await?;
                let elapsed = start.elapsed();
                let ms = elapsed.as_secs_f64() * 1000.0;
                millis.push(ms);
                let row_count: usize = results.iter().map(|b| b.num_rows()).sum();
                println!(
                    "Query {query_id} iteration {i} took {ms:.1} ms and returned {row_count} rows"
                );
                benchmark_run.write_iter(elapsed, row_count);
            }
            let avg = millis.iter().sum::<f64>() / millis.len() as f64;
            println!("Query {query_id} avg time: {avg:.2} ms");

            // Print memory usage stats using mimalloc (only when compiled with --features mimalloc_extended)
            print_memory_stats();