feat: add api to execute clickbench benchmark

use API `/api/v1/benchmark/clickbench`
to perform benchmark on clickbench dataset and defined queries

add env `PARQUET_FILE` to provide file path of hits.parquet
add env `QUERIES_FILE` to provide file path of queries file

3 tries for each query, total 43 queries in the set
api response with query no, iteration no, response time in ms

Author: nikhilsinhaparseable

src/handlers/http/clickbench.rs

@@ -0,0 +1,123 @@
+/*
+ * Parseable Server (C) 2022 - 2024 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use std::{collections::HashMap, env, fs, time::Instant};
+
+use actix_web::{web::Json, Responder};
+use datafusion::{
+    common::plan_datafusion_err,
+    error::DataFusionError,
+    execution::{runtime_env::RuntimeEnvBuilder, SessionStateBuilder},
+    physical_plan::collect,
+    prelude::{ParquetReadOptions, SessionConfig, SessionContext},
+    sql::{parser::DFParser, sqlparser::dialect::dialect_from_str},
+};
+use serde_json::{json, Value};
+
+pub async fn clickbench_benchmark() -> Result<impl Responder, actix_web::Error> {
+    let results = tokio::task::spawn_blocking(run_benchmark)
+        .await
+        .map_err(actix_web::error::ErrorInternalServerError)?
+        .map_err(actix_web::error::ErrorInternalServerError)?;
+    Ok(results)
+}
+
+#[tokio::main(flavor = "multi_thread")]
+pub async fn run_benchmark() -> Result<Json<Value>, anyhow::Error> {
+    let mut session_config = SessionConfig::from_env()?.with_information_schema(true);
+
+    session_config = session_config.with_batch_size(8192);
+
+    let rt_builder = RuntimeEnvBuilder::new();
+    // set memory pool size
+    let runtime_env = rt_builder.build_arc()?;
+    let state = SessionStateBuilder::new()
+        .with_default_features()
+        .with_config(session_config)
+        .with_runtime_env(runtime_env)
+        .build();
+    state
+        .catalog_list()
+        .catalog(&state.config_options().catalog.default_catalog)
+        .expect("default catalog is provided by datafusion");
+
+    let ctx = SessionContext::new_with_state(state);
+
+    let mut table_options = HashMap::new();
+    table_options.insert("binary_as_string", "true");
+
+    let parquet_file = env::var("PARQUET_FILE")?;
+    register_hits(&ctx, &parquet_file).await?;
+    let mut query_list = Vec::new();
+    let queries_file = env::var("QUERIES_FILE")?;
+    let queries = fs::read_to_string(queries_file)?;
+    for query in queries.lines() {
+        query_list.push(query.to_string());
+    }
+    execute_queries(&ctx, query_list).await
+}
+
+async fn register_hits(ctx: &SessionContext, parquet_file: &str) -> Result<(), anyhow::Error> {
+    let options: ParquetReadOptions<'_> = Default::default();
+    ctx.register_parquet("hits", parquet_file, options)
+        .await
+        .map_err(|e| {
+            DataFusionError::Context(format!("Registering 'hits' as {parquet_file}"), Box::new(e))
+        })?;
+    Ok(())
+}
+
+pub async fn execute_queries(
+    ctx: &SessionContext,
+    query_list: Vec<String>,
+) -> Result<Json<Value>, anyhow::Error> {
+    const TRIES: usize = 3;
+    let mut results = Vec::new();
+
+    for sql in query_list.iter() {
+        let mut elapsed_times = Vec::new();
+        for _iteration in 1..=TRIES {
+            let start = Instant::now();
+            let task_ctx = ctx.task_ctx();
+            let dialect = &task_ctx.session_config().options().sql_parser.dialect;
+            let dialect = dialect_from_str(dialect).ok_or_else(|| {
+                plan_datafusion_err!(
+                    "Unsupported SQL dialect: {dialect}. Available dialects: \
+                     Generic, MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, \
+                     MsSQL, ClickHouse, BigQuery, Ansi."
+                )
+            })?;
+
+            let statements = DFParser::parse_sql_with_dialect(sql, dialect.as_ref())?;
+            let statement = statements.front().unwrap();
+            let plan = ctx.state().statement_to_plan(statement.clone()).await?;
+
+            let df = ctx.execute_logical_plan(plan).await?;
+            let physical_plan = df.create_physical_plan().await?;
+
+            let _ = collect(physical_plan, task_ctx.clone()).await?;
+            let elapsed = start.elapsed().as_secs_f64();
+            elapsed_times.push(elapsed);
+        }
+        results.push(elapsed_times);
+    }
+
+    let result_json = json!(results);
+
+    Ok(Json(result_json))
+}

src/handlers/http/mod.rs

@@ -30,6 +30,7 @@ use self::{cluster::get_ingestor_info, query::Query};
 pub mod about;
 pub mod alerts;
 mod audit;
+pub mod clickbench;
 pub mod cluster;
 pub mod correlation;
 pub mod health_check;

src/handlers/http/modal/server.rs

@@ -25,6 +25,7 @@ use crate::handlers;
 use crate::handlers::http::about;
 use crate::handlers::http::alerts;
 use crate::handlers::http::base_path;
+use crate::handlers::http::clickbench;
 use crate::handlers::http::health_check;
 use crate::handlers::http::query;
 use crate::handlers::http::users::dashboards;
@@ -87,7 +88,8 @@ impl ParseableServer for Server {
                     .service(Self::get_user_role_webscope())
                     .service(Self::get_counts_webscope())
                     .service(Self::get_alerts_webscope())
-                    .service(Self::get_metrics_webscope()),
+                    .service(Self::get_metrics_webscope())
+                    .service(Self::get_benchmark_webscope()),
             )
             .service(Self::get_ingest_otel_factory())
             .service(Self::get_generated());
@@ -160,6 +162,16 @@ impl Server {
         )
     }
 
+    pub fn get_benchmark_webscope() -> Scope {
+        web::scope("/benchmark/clickbench").service(
+            web::resource("").route(
+                web::get()
+                    .to(clickbench::clickbench_benchmark)
+                    .authorize(Action::Benchmark),
+            ),
+        )
+    }
+
     pub fn get_correlation_webscope() -> Scope {
         web::scope("/correlation")
             .service(

src/parseable/streams.rs

@@ -513,10 +513,7 @@ impl Stream {
                     let file_size = match file.metadata() {
                         Ok(meta) => meta.len(),
                         Err(err) => {
-                            warn!(
-                                "File ({}) not found; Error = {err}",
-                                file.display()
-                            );
+                            warn!("File ({}) not found; Error = {err}", file.display());
                             continue;
                         }
                     };

src/rbac/role.rs

@@ -67,6 +67,7 @@ pub enum Action {
     CreateCorrelation,
     DeleteCorrelation,
     PutCorrelation,
+    Benchmark,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -108,6 +109,7 @@ impl RoleBuilder {
                 ),
                 Action::Login
                 | Action::Metrics
+                | Action::Benchmark
                 | Action::PutUser
                 | Action::ListUser
                 | Action::PutUserRoles