feat: Update testoperator dispatch for validation, version metric (spiceai#5349)

* feat: Update testoperator dispatch for validation, version metric

* fix: Spiced version in test framework

* fix: Print stderr if test framework fails spiced version

* chore: Clippy

Author: peasee

.github/workflows/testoperator_dispatch.yml

@@ -29,6 +29,11 @@ on:
         required: false
         type: boolean
         default: false
+      validate:
+        description: 'Validate results?'
+        required: false
+        type: boolean
+        default: false
 
 jobs:
   dispatch-tests:
@@ -62,7 +67,8 @@ jobs:
           testoperator dispatch \
             ${{ github.event.inputs.test_files_path }} \
             --workflow ${{ github.event.inputs.workflow_type }} \
-            --update-snapshots ${{ github.event.inputs.update_snapshots }}
+            --update-snapshots ${{ github.event.inputs.update_snapshots }} \
+            --validate=${{ github.event.inputs.validate }} \
         env:
           GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SPICED_COMMIT: ${{ steps.setup-spiced.outputs.SPICED_COMMIT }}

crates/test-framework/src/metrics/mod.rs

@@ -293,6 +293,7 @@ impl StatisticsCollector<BTreeMap<String, Duration>, BTreeMap<String, Vec<Durati
 pub struct QueryMetrics<T: ExtendedMetrics, R: ExtendedMetrics> {
     pub run_id: Uuid,
     pub run_name: String,
+    pub spiced_version: String,
     pub commit_sha: String,
     pub branch_name: String,
     pub test_type: TestType,
@@ -373,6 +374,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
 
         let mut base_fields = vec![
             Field::new("run_id", DataType::Utf8, false),
+            Field::new("spiced_version", DataType::Utf8, false),
             Field::new("run_name", DataType::Utf8, false),
             Field::new("commit_sha", DataType::Utf8, false),
             Field::new("branch_name", DataType::Utf8, false),
@@ -397,6 +399,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
 
         let mut base_fields = vec![
             Field::new("run_id", DataType::Utf8, false),
+            Field::new("spiced_version", DataType::Utf8, false),
             Field::new("started_at", DataType::Int64, false),
             Field::new("finished_at", DataType::Int64, false),
             Field::new("query_name", DataType::Utf8, false),
@@ -499,6 +502,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
     #[allow(clippy::cast_possible_wrap)]
     pub fn build_records(&self) -> Result<Vec<RecordBatch>> {
         let run_id = vec![self.run_id.to_string(); self.metrics.len()];
+        let spiced_version = vec![self.spiced_version.clone(); self.metrics.len()];
 
         let started_at = extract_metric_values!(self.metrics, started_at, as_i64);
         let finished_at = extract_metric_values!(self.metrics, finished_at, as_i64);
@@ -520,6 +524,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
 
         let mut columns: Vec<ArrayRef> = vec![
             Arc::new(StringArray::from(run_id)),
+            Arc::new(StringArray::from(spiced_version)),
             Arc::new(Int64Array::from(started_at)),
             Arc::new(Int64Array::from(finished_at)),
             Arc::new(StringArray::from(query_name)),
@@ -569,6 +574,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
     /// The record batch is a single row, representing the run as a whole - which can pass or fail separately from individual queries
     pub fn build_run(&self, status: QueryStatus) -> Result<Vec<RecordBatch>> {
         let run_id = vec![self.run_id.to_string()];
+        let spiced_version = vec![self.spiced_version.to_string()];
         let run_name = vec![self.run_name.clone()];
         let commit_sha = vec![self.commit_sha.clone()];
         let branch_name = vec![self.branch_name.clone()];
@@ -602,6 +608,7 @@ impl<T: ExtendedMetrics, R: ExtendedMetrics> QueryMetrics<T, R> {
 
         let mut columns: Vec<ArrayRef> = vec![
             Arc::new(StringArray::from(run_id)),
+            Arc::new(StringArray::from(spiced_version)),
             Arc::new(StringArray::from(run_name)),
             Arc::new(StringArray::from(commit_sha)),
             Arc::new(StringArray::from(branch_name)),
@@ -650,11 +657,13 @@ pub trait MetricCollector<T: ExtendedMetrics, R: ExtendedMetrics> {
     fn start_time(&self) -> SystemTime;
     fn end_time(&self) -> SystemTime;
     fn name(&self) -> String;
+    fn spiced_version(&self) -> Result<&str>;
     fn metrics(&self) -> Result<Vec<QueryMetric<T>>>;
     fn collect(&self, test_type: TestType) -> Result<QueryMetrics<T, R>> {
         Ok(QueryMetrics {
             run_id: uuid::Uuid::new_v4(),
             run_name: self.name(),
+            spiced_version: self.spiced_version()?.to_string(),
             commit_sha: git::get_commit_sha(),
             branch_name: git::get_branch_name(),
             test_type,

crates/test-framework/src/spiced/mod.rs

@@ -15,6 +15,7 @@ limitations under the License.
 */
 
 use std::{
+    fmt::Display,
     path::PathBuf,
     process::{Child, Command},
     time::Duration,
@@ -28,9 +29,25 @@ use tempfile::TempDir;
 
 use crate::{process::Process, utils::wait_until_true};
 
+#[derive(Debug, Clone)]
+pub struct SpicedVersion(String);
+impl SpicedVersion {
+    #[must_use]
+    pub fn new(version: String) -> Self {
+        Self(version)
+    }
+}
+
+impl Display for SpicedVersion {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 pub struct SpicedInstance {
     child: Child,
     tempdir: TempDir,
+    version: SpicedVersion,
 }
 
 pub struct StartRequest {
@@ -113,13 +130,42 @@ impl SpicedInstance {
 
         let tempdir = start_request.tempdir;
 
+        // Get spiced version
+        let version_cmd = Command::new(start_request.spiced_path.clone())
+            .arg("--version")
+            .output()?;
+
+        if !version_cmd.status.success() {
+            anyhow::bail!(
+                "Failed to get spiced version: {}",
+                String::from_utf8_lossy(&version_cmd.stderr)
+            );
+        }
+
+        let version = String::from_utf8_lossy(&version_cmd.stdout).to_string();
+        // take just the v1.0.0 part of the version
+        let version = match (version.contains('-'), version.contains('+')) {
+            (true, _) => version.split('-').next().unwrap_or(&version).to_string(),
+            (false, true) => version.split('+').next().unwrap_or(&version).to_string(),
+            (false, false) => version,
+        };
+
         // Start the spiced instance
         let mut cmd = Command::new(start_request.spiced_path);
         cmd.current_dir(tempdir.path());
         cmd.arg("--telemetry-enabled=false");
         let child = cmd.spawn()?;
 
-        Ok(Self { child, tempdir })
+        Ok(Self {
+            child,
+            tempdir,
+            version: SpicedVersion::new(version),
+        })
+    }
+
+    #[must_use]
+    pub fn version(&self) -> &str {
+        self.version.0.as_str()
     }
 
     #[must_use]

crates/test-framework/src/spicetest/datasets/mod.rs

@@ -384,6 +384,15 @@ impl MetricCollector<DatasetMetrics, NoExtendedMetrics> for SpiceTest<Completed>
         self.name.clone()
     }
 
+    fn spiced_version(&self) -> Result<&str> {
+        let spiced_instance = self.spiced_instance.as_ref().ok_or(
+            anyhow::anyhow!(
+                "Spiced instance is not available. SpiceTest must be started before metrics can be collected."
+            ))?;
+
+        Ok(spiced_instance.version())
+    }
+
     fn metrics(&self) -> Result<Vec<QueryMetric<DatasetMetrics>>> {
         self.get_query_durations()
             .iter()
@@ -429,6 +438,15 @@ impl MetricCollector<NoExtendedMetrics, ThroughputMetrics> for SpiceTest<Complet
         self.name.clone()
     }
 
+    fn spiced_version(&self) -> Result<&str> {
+        let spiced_instance = self.spiced_instance.as_ref().ok_or(
+            anyhow::anyhow!(
+                "Spiced instance is not available. SpiceTest must be started before metrics can be collected."
+            ))?;
+
+        Ok(spiced_instance.version())
+    }
+
     fn metrics(&self) -> Result<Vec<QueryMetric<NoExtendedMetrics>>> {
         self.get_query_durations()
             .iter()

crates/test-framework/src/spicetest/http/consistency.rs

@@ -212,6 +212,15 @@ impl MetricCollector<NoExtendedMetrics, NoExtendedMetrics> for SpiceTest<Complet
         self.name.clone()
     }
 
+    fn spiced_version(&self) -> Result<&str> {
+        let spiced_instance = self.spiced_instance.as_ref().ok_or(
+            anyhow::anyhow!(
+                "Spiced instance is not available. SpiceTest must be started before metrics can be collected."
+            ))?;
+
+        Ok(spiced_instance.version())
+    }
+
     fn metrics(&self) -> Result<Vec<QueryMetric<NoExtendedMetrics>>> {
         self.state
             .result

crates/test-framework/src/spicetest/http/overhead.rs

@@ -193,6 +193,15 @@ impl MetricCollector<NoExtendedMetrics, NoExtendedMetrics> for SpiceTest<Complet
         self.name.clone()
     }
 
+    fn spiced_version(&self) -> Result<&str> {
+        let spiced_instance = self.spiced_instance.as_ref().ok_or(
+            anyhow::anyhow!(
+                "Spiced instance is not available. SpiceTest must be started before metrics can be collected."
+            ))?;
+
+        Ok(spiced_instance.version())
+    }
+
     fn metrics(&self) -> Result<Vec<QueryMetric<NoExtendedMetrics>>> {
         let baseline = QueryMetric::new_from_durations(
             "baseline",

crates/test-framework/src/spicetest/vector_search/mod.rs

@@ -149,6 +149,15 @@ impl MetricCollector<SearchScoreMetric, NoExtendedMetrics> for SpiceTest<Complet
         self.name.clone()
     }
 
+    fn spiced_version(&self) -> Result<&str> {
+        let spiced_instance = self.spiced_instance.as_ref().ok_or(
+            anyhow::anyhow!(
+                "Spiced instance is not available. SpiceTest must be started before metrics can be collected."
+            ))?;
+
+        Ok(spiced_instance.version())
+    }
+
     fn metrics(&self) -> Result<Vec<QueryMetric<SearchScoreMetric>>> {
         self.state
             .results

tools/testoperator/dispatch/tpch/accelerated/file_arrow.yaml

@@ -3,6 +3,7 @@ tests:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_arrow.yaml
     query_set: tpch
     runner_type: spiceai-runners
+    validate: true
   throughput:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_arrow.yaml
     query_set: tpch

tools/testoperator/dispatch/tpch/accelerated/file_duckdb_file.yaml

@@ -3,6 +3,7 @@ tests:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_duckdb_file.yaml
     query_set: tpch
     runner_type: spiceai-runners
+    validate: true
   throughput:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_duckdb_file.yaml
     query_set: tpch

tools/testoperator/dispatch/tpch/accelerated/file_duckdb_memory.yaml

@@ -3,6 +3,7 @@ tests:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_duckdb_memory.yaml
     query_set: tpch
     runner_type: spiceai-runners
+    validate: true
   throughput:
     spicepod_path: ./test/spicepods/tpch/accelerated/file_duckdb_memory.yaml
     query_set: tpch