Add Scenario enum and simplify load runner to not reference spiced directly

Introduce a Scenario enum that encapsulates benchmark configuration (query set,
end condition) and use it to drive the load runner instead of coupling to
BenchRunArgs/DatasetTestArgs and direct spiced instance management.

- Add src/scenario.rs with Scenario enum, Display, load_query_set(), and
  end_condition() (TPCH runs for 60s)
- Simplify load runner to accept Scenario + CommonArgs + ADBC connection,
  removing warm-up/baseline phases, spiced lifecycle, memory monitoring,
  health monitoring, and query overrides/validation
- Remove health.rs (HealthMonitor) as it depended on direct spiced access
- Trim CommonArgs to scenario-relevant fields (remove spicepod_path,
  spiced_path, duration, data_dir, etc.)
- Simplify build_test_with_validation to take Scenario instead of
  DatasetTestArgs
- Clean up unused imports, re-exports, and clippy warnings

Author: phillipleblanc

src/args/mod.rs

@@ -14,52 +14,22 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-use std::path::PathBuf;
-
 use clap::{ArgAction, Parser, ValueEnum};
 
 mod dataset;
-pub use dataset::{BenchRunArgs, DatasetTestArgs};
+use crate::scenario::Scenario;
 
 /// Arguments Common to all [`TestCommands`].
 #[derive(Parser, Debug, Clone)]
 pub struct CommonArgs {
-    /// Path to the spicepod.yaml file
-    #[arg(short('p'), long, default_value = "spicepod.yaml")]
-    pub(crate) spicepod_path: PathBuf,
-
-    #[arg(short('z'), long)]
-    pub(crate) spicepod_dependencies: Option<PathBuf>,
+    /// The scenario to use for the benchmark run, which determines the query set and other parameters.
+    #[arg(long)]
+    pub(crate) scenario: Scenario,
 
     /// The number of clients to run simultaneously. Each client will send a query, wait for a response, then send another query.
     #[arg(long, default_value = "1")]
     pub(crate) concurrency: usize,
 
-    /// Path to the spiced binary, or URL to an already-running spiced instance's Flight endpoint
-    /// (e.g., `http://localhost:50051` to connect to an external instance)
-    #[arg(short, long, default_value = "spiced")]
-    pub(crate) spiced_path: String,
-
-    /// The number of seconds to wait for the spiced instance to become ready
-    #[arg(long, default_value = "30")]
-    pub(crate) ready_wait: u64,
-
-    /// The duration of the test in seconds
-    #[arg(long, default_value = "60")]
-    pub(crate) duration: u64,
-
-    /// Whether to disable progress bars, for CI or non-interactive environments
-    #[arg(long)]
-    pub(crate) disable_progress_bars: bool,
-
-    /// An optional data directory, to symlink into the spiced instance
-    #[arg(short, long)]
-    pub(crate) data_dir: Option<PathBuf>,
-
-    /// Whether to enable metrics collection
-    #[arg(long)]
-    pub(crate) metrics: bool,
-
     /// Whether to collect SUT metrics via the system adapter JSON-RPC command.
     #[arg(long)]
     pub(crate) scrape_sut_metrics: bool,
@@ -108,23 +78,6 @@ pub struct CommonArgs {
     pub(crate) system_adapter_env: Vec<(String, String)>,
 }
 
-#[allow(dead_code)]
-impl CommonArgs {
-    /// Check if `spiced_path` is a URL to an external instance
-    #[must_use]
-    #[allow(dead_code)]
-    pub fn is_external_instance(&self) -> bool {
-        self.spiced_path.starts_with("http://") || self.spiced_path.starts_with("https://")
-    }
-
-    /// Get the spiced path as a `PathBuf` (only valid when not an external instance)
-    #[must_use]
-    #[allow(dead_code)]
-    pub fn spiced_path_buf(&self) -> PathBuf {
-        PathBuf::from(&self.spiced_path)
-    }
-}
-
 fn parse_key_val(s: &str) -> Result<(String, String), String> {
     let pos = s
         .find('=')

src/commands/load/mod.rs

@@ -15,20 +15,18 @@ limitations under the License.
 */
 #![allow(dead_code)]
 
-use crate::args::BenchRunArgs;
+use crate::{args::CommonArgs, commands::adbc_executor, scenario::Scenario};
 use std::sync::Arc;
 use std::time::Duration;
 use system_adapter_protocol::MetricsResponse;
 use test_framework::{
     TestType, anyhow,
     arrow::util::pretty::print_batches,
-    git,
     metrics::{MetricCollector, NoExtendedMetrics, QueryMetrics, QueryStatus, StatisticsCollector},
     opentelemetry::KeyValue,
-    opentelemetry_sdk::Resource,
     spicetest::{
         SpiceTest,
-        datasets::{EndCondition, NotStarted},
+        datasets::NotStarted,
     },
     telemetry::streaming::StreamingOtlpExporter,
 };
@@ -115,82 +113,31 @@ fn spawn_sut_metrics_scraper(
 
 #[expect(clippy::too_many_lines)]
 pub(crate) async fn run(
-    args: &BenchRunArgs,
-    adbc_conn: Option<adbc_client::AdbcConnection>,
+    scenario: &Scenario,
+    common_args: &CommonArgs,
+    adbc_conn: adbc_client::AdbcConnection,
 ) -> anyhow::Result<()> {
-    if args.test_args.common.concurrency < 2 {
-        return Err(anyhow::anyhow!(
-            "Concurrency should be greater than 1 for a load test"
-        ));
-    }
-
-    let sut_name = "spicebench-sut";
-
-    let spiced_commit_sha =
-        std::env::var("SPICED_COMMIT").unwrap_or_else(|_| "unknown".to_string());
-    let spicebench_commit_sha = git::get_commit_sha();
-    let branch_name = git::get_branch_name();
-    let spicepod = args.test_args.common.spicepod_path.display().to_string();
-
-    let query_set = args.test_args.load_query_set()?;
-    let load_resource = Resource::builder_empty()
-        .with_attributes(vec![
-            KeyValue::new("service.name", sut_name.to_string()),
-            KeyValue::new("type", "spicebench"),
-            KeyValue::new("name", sut_name),
-            KeyValue::new("query_set", query_set.to_string()),
-            KeyValue::new("spicebench_commit_sha", spicebench_commit_sha),
-            KeyValue::new("spiced_commit_sha", spiced_commit_sha),
-            KeyValue::new("branch_name", branch_name),
-            KeyValue::new("concurrency", args.test_args.common.concurrency.to_string()),
-            KeyValue::new("spicepod", spicepod),
-            KeyValue::new(
-                "param_set_variants",
-                args.test_args
-                    .random_param_set_count
-                    .unwrap_or(1)
-                    .to_string(),
-            ),
-            KeyValue::new(
-                "protocol",
-                if args.test_args.http_clients {
-                    "http"
-                } else {
-                    "flight"
-                },
-            ),
-        ])
-        .build();
-
-    // Create telemetry with resource upfront, before any metrics calls
-    let telemetry = super::create_telemetry_with_resource(&args.test_args.common, load_resource);
-
+    scenario.load_query_set()?;
     // Create the appropriate query executor based on args
-    let executor = super::create_query_executor(&args.test_args, None, adbc_conn).await?;
+    let executor = Box::new(adbc_executor::AdbcDirectQueryExecutor::new(adbc_conn));
 
-    println!("Starting Spicebench run");
+    println!("Running benchmark");
 
-    let load_end_condition = if args.run_until_stopped {
-        EndCondition::Unlimited
-    } else {
-        EndCondition::Duration(Duration::from_secs(args.test_args.common.duration))
-    };
+    let load_end_condition = scenario.end_condition();
 
     // Create streaming OTLP exporter if OTLP endpoint is configured
-    let streaming_exporter = args
-        .test_args
-        .common
+    let streaming_exporter = common_args
         .otlp_endpoint
         .as_ref()
         .map(|endpoint| StreamingOtlpExporter::spawn(endpoint.clone()));
 
     // Spawn SUT metrics scraper if --scrape-sut-metrics is enabled and a system adapter is configured
     let sut_scraper_token = CancellationToken::new();
-    let sut_scraper_handle = if args.test_args.common.scrape_sut_metrics
-        && (args.test_args.common.system_adapter_stdio_cmd.is_some()
-            || args.test_args.common.system_adapter_http_url.is_some())
+    let sut_scraper_handle = if common_args.scrape_sut_metrics
+        && (common_args.system_adapter_stdio_cmd.is_some()
+            || common_args.system_adapter_http_url.is_some())
     {
-        let adapter = super::connect_system_adapter(&args.test_args.common).await?;
+        let adapter = super::connect_system_adapter(common_args).await?;
         let run_id = uuid::Uuid::new_v4();
         println!("SUT metrics scraping enabled (run_id={run_id})");
         Some(spawn_sut_metrics_scraper(
@@ -204,43 +151,33 @@ pub(crate) async fn run(
     };
 
     // Record client concurrency as a gauge
-    crate::metrics::ACTIVE_CONNECTIONS.record(
-        args.test_args.common.concurrency.try_into().unwrap_or(0),
-        &[],
-    );
+    crate::metrics::ACTIVE_CONNECTIONS.record(common_args.concurrency.try_into().unwrap_or(0), &[]);
 
     let mut test_builder = NotStarted::new()
-        .with_parallel_count(args.test_args.common.concurrency)
+        .with_parallel_count(common_args.concurrency)
         .with_end_condition(load_end_condition)
-        .with_query_executor(executor)
-        .with_query_duration_threshold(args.test_args.mark_query_failed_if_exceeds);
+        .with_query_executor(executor);
 
     // Add streaming metrics sender if exporter is configured
     if let Some(exporter) = &streaming_exporter {
         test_builder = test_builder.with_streaming_metrics(exporter.sender());
     }
 
     let (query_set, test_builder) =
-        super::build_test_with_validation(&args.test_args, test_builder).await?;
+        super::build_test_with_validation(scenario, test_builder).await?;
 
-    // Use the same query overrides that were applied in build_test_with_validation
-    let query_overrides = args
-        .test_args
-        .query_overrides
-        .clone()
-        .map(test_framework::queries::QueryOverrides::from);
-    let _queries = query_set.get_queries(query_overrides, None, None).await?;
+    let _queries = query_set.get_queries(None, None, None).await?;
 
-    let throughput_test = SpiceTest::<NotStarted>::new("spicebench".into(), test_builder)
-        .with_progress_bars(!args.test_args.common.disable_progress_bars)
+    let throughput_test = SpiceTest::<NotStarted>::new(scenario.to_string(), test_builder)
+        .with_progress_bars(false)
         .start()?;
     let shutdown_token = throughput_test.cancellation_token();
     let test_future = throughput_test.wait();
     tokio::pin!(test_future);
     let test = match tokio::select! {
         res = &mut test_future => res,
         _ = signal::ctrl_c() => {
-            println!("Interrupt received, stopping load test...");
+            println!("Interrupt received, stopping benchmark...");
             shutdown_token.cancel();
             test_future.await
         }
@@ -250,11 +187,14 @@ pub(crate) async fn run(
             return Err(e);
         }
     };
+    test.get_query_durations().statistical_set()?;
+
     // Get all query durations for overall statistics before ending the test
     let all_durations = test.get_query_durations().clone();
     let all_duration_values: Vec<_> = all_durations.values().flatten().copied().collect();
 
     let metrics: QueryMetrics<_, NoExtendedMetrics> = test.collect(TestType::Load)?;
+    let _ = test.end();
 
     // Record per-query metrics for load test
     for query in &metrics.metrics {
@@ -283,13 +223,8 @@ pub(crate) async fn run(
         .record((metrics.finished_at - metrics.started_at).try_into()?, &[]);
 
     // Query throughput metrics
-    let total_iterations: u64 = metrics
-        .metrics
-        .iter()
-        .map(|q| q.iterations as u64)
-        .sum();
-    let test_duration_secs =
-        (metrics.finished_at - metrics.started_at) as f64 / 1000.0;
+    let total_iterations: u64 = metrics.metrics.iter().map(|q| q.iterations as u64).sum();
+    let test_duration_secs = (metrics.finished_at - metrics.started_at) as f64 / 1000.0;
     crate::metrics::QUERIES_TOTAL.add(total_iterations, &[]);
     if test_duration_secs > 0.0 {
         let qps = total_iterations as f64 / test_duration_secs;
@@ -300,8 +235,7 @@ pub(crate) async fn run(
             .map(|n| n.get() as f64)
             .unwrap_or(1.0);
         if cpu_cores > 0.0 {
-            crate::metrics::EFFICIENCY_QUERIES_PER_CORE
-                .record(qps / cpu_cores, &[]);
+            crate::metrics::EFFICIENCY_QUERIES_PER_CORE.record(qps / cpu_cores, &[]);
         }
     }
 
@@ -319,7 +253,8 @@ pub(crate) async fn run(
         );
     }
 
-    println!("Load test metrics:");
+    println!("{}", vec!["-"; 30].join(""));
+    println!("Benchmark metrics:");
     let records = metrics.build_records()?;
     print_batches(&records)?;
 
@@ -328,9 +263,6 @@ pub(crate) async fn run(
         exporter.shutdown().await;
     }
 
-    telemetry.emit().await?;
-
-    println!("Spicebench run completed");
-
+    println!("Benchmark completed");
     Ok(())
 }