Method benchmark_query

benchmarks/src/bin/external_aggr.rs:185–240

Benchmark query `query_id` in `AGGR_QUERIES`

(
        &self,
        query_id: usize,
        mem_limit: u64,
        mem_pool_type: &str,
    )

    /// Benchmark query `query_id` in `AGGR_QUERIES`
    async fn benchmark_query(
        &self,
        query_id: usize,
        mem_limit: u64,
        mem_pool_type: &str,
    ) -> Result<Vec<QueryResult>> {
        let query_name =
            format!("Q{query_id}({})", human_readable_size(mem_limit as usize));
        let config = self.common.config()?;
        let memory_pool: Arc<dyn MemoryPool> = match mem_pool_type {
            "fair" => Arc::new(FairSpillPool::new(mem_limit as usize)),
            "greedy" => Arc::new(GreedyMemoryPool::new(mem_limit as usize)),
            _ => {
                return exec_err!("Invalid memory pool type: {}", mem_pool_type);
            }
        };
        let runtime_env = RuntimeEnvBuilder::new()
            .with_memory_pool(memory_pool)
            .build_arc()?;
        let state = SessionStateBuilder::new()
            .with_config(config)
            .with_runtime_env(runtime_env)
            .with_default_features()
            .build();
        let ctx = SessionContext::from(state);

        // register tables
        self.register_tables(&ctx).await?;

        let mut millis = vec![];
        // run benchmark
        let mut query_results = vec![];
        for i in 0..self.iterations() {
            let start = Instant::now();

            let query_idx = query_id - 1; // 1-indexed -> 0-indexed
            let sql = Self::AGGR_QUERIES[query_idx];

            let result = self.execute_query(&ctx, sql).await?;

            let elapsed = start.elapsed(); //.as_secs_f64() * 1000.0;
            let ms = elapsed.as_secs_f64() * 1000.0;
            millis.push(ms);

            let row_count = result.iter().map(|b| b.num_rows()).sum();
            println!(
                "{query_name} iteration {i} took {ms:.1} ms and returned {row_count} rows"
            );
            query_results.push(QueryResult { elapsed, row_count });
        }

        let avg = millis.iter().sum::<f64>() / millis.len() as f64;
        println!("{query_name} avg time: {avg:.2} ms");

        Ok(query_results)
    }

    async fn register_tables(&self, ctx: &SessionContext) -> Result<()> {