//! Selector API for returning subset of series which will be rendered in some
//! format.
//!
//! We have the following expected paths:
//!
//! * :benchmark/:profile/:scenario/:metric => [cid => u64]
//! * :crate/:profile/:scenario/:self_profile_query/:stat (SelfProfileTime, SelfProfileCacheHits, ...)
//!   :stat = time => Duration,
//!   :stat = cache hits => u32,
//!   :stat = invocation count => u32,
//!   :stat = blocked time => Duration,
//!   :stat = incremental load time => Duration,
//!
//! Note that the returned series always have a "simple" type of a small set --
//! things like arrays, integers. We aggregate into higher level types above the
//! primitive series readers.
//!
//! We specify a single struct per path style above.
//!
//! `Option<T>` in the path either specifies a specific T to filter by, or
//! requests that all are provided. Note that this is a cartesian product if
//! there are multiple `None`s.

use std::{
    fmt::{self, Debug},
    hash::Hash,
    sync::Arc,
};

use crate::{
    interpolate::Interpolate, metric::Metric, ArtifactId, ArtifactIdIter, Benchmark,
    CodegenBackend, Connection, Index, Profile, Scenario, Tag, Target,
};

use serde::{Deserialize, Serialize};

/// How to aggregate multiple samples for the same (artifact, test case) pair.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default, PartialEq, Eq, Hash)]
#[serde(rename_all = "snake_case")]
pub enum Aggregation {
    #[default]
    Min,
    ArithmeticMean,
    GeometricMean,
}

/// Aggregate a slice of samples into a single value using the given method.
/// Returns `None` if the slice is empty.
///
/// For arithmetic and geometric mean, if there are more than 3 samples the
/// single largest and single smallest values are trimmed before computing the
/// mean (trimmed mean) to reduce the impact of outliers.
pub fn aggregate(samples: &[f64], method: Aggregation) -> Option<f64> {
    if samples.is_empty() {
        return None;
    }
    Some(match method {
        Aggregation::Min => samples
            .iter()
            .copied()
            .min_by(|a, b| a.total_cmp(b))
            .unwrap(),
        Aggregation::ArithmeticMean => {
            let trimmed = trim_outliers(samples);
            trimmed.iter().sum::<f64>() / trimmed.len() as f64
        }
        Aggregation::GeometricMean => {
            let trimmed = trim_outliers(samples);
            let log_sum: f64 = trimmed.iter().map(|x| x.ln()).sum();
            (log_sum / trimmed.len() as f64).exp()
        }
    })
}

/// If `samples` has more than 3 elements, return a new vec with the single
/// smallest and single largest values removed.  Otherwise return the original
/// slice as a vec (unchanged).
fn trim_outliers(samples: &[f64]) -> Vec<f64> {
    if samples.len() <= 3 {
        return samples.to_vec();
    }
    let mut sorted = samples.to_vec();
    sorted.sort_by(|a, b| a.total_cmp(b));
    // Remove first (min) and last (max).
    sorted[1..sorted.len() - 1].to_vec()
}

#[derive(Debug)]
pub struct StatisticSeries {
    pub artifact_ids: ArtifactIdIter,
    pub points: std::vec::IntoIter<Option<f64>>,
}

impl Iterator for StatisticSeries {
    type Item = (ArtifactId, Option<f64>);
    fn next(&mut self) -> Option<Self::Item> {
        Some((self.artifact_ids.next()?, self.points.next().unwrap()))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.artifact_ids.size_hint()
    }
}

pub trait Point {
    type Key: fmt::Debug + PartialEq + Clone;

    fn key(&self) -> &Self::Key;
    fn set_key(&mut self, key: Self::Key);
    fn value(&self) -> Option<f64>;
    fn set_value(&mut self, value: f64);
    fn interpolated(&self) -> bool;
    fn set_interpolated(&mut self);
}

impl<T: Clone + PartialEq + fmt::Debug> Point for (T, Option<f64>) {
    type Key = T;

    fn key(&self) -> &T {
        &self.0
    }
    fn set_key(&mut self, key: T) {
        self.0 = key;
    }
    fn value(&self) -> Option<f64> {
        self.1
    }
    fn set_value(&mut self, value: f64) {
        self.1 = Some(value);
    }
    fn interpolated(&self) -> bool {
        false
    }
    fn set_interpolated(&mut self) {
        // no-op
    }
}

impl<T: Clone + PartialEq + fmt::Debug> Point for (T, f64) {
    type Key = T;

    fn key(&self) -> &T {
        &self.0
    }
    fn set_key(&mut self, key: T) {
        self.0 = key;
    }
    fn value(&self) -> Option<f64> {
        Some(self.1)
    }
    fn set_value(&mut self, value: f64) {
        self.1 = value;
    }
    fn interpolated(&self) -> bool {
        false
    }
    fn set_interpolated(&mut self) {
        // no-op
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum Selector<T> {
    All,
    Subset(Vec<T>),
    One(T),
}

impl<T> Selector<T> {
    fn map<U>(self, mut f: impl FnMut(T) -> U) -> Selector<U> {
        match self {
            Selector::All => Selector::All,
            Selector::Subset(subset) => Selector::Subset(subset.into_iter().map(f).collect()),
            Selector::One(o) => Selector::One(f(o)),
        }
    }
    pub fn try_map<U, E>(self, mut f: impl FnMut(T) -> Result<U, E>) -> Result<Selector<U>, E> {
        Ok(match self {
            Selector::All => Selector::All,
            Selector::Subset(subset) => {
                Selector::Subset(subset.into_iter().map(f).collect::<Result<_, _>>()?)
            }
            Selector::One(o) => Selector::One(f(o)?),
        })
    }

    fn matches<U>(&self, other: U) -> bool
    where
        U: PartialEq<T>,
    {
        match self {
            Selector::One(c) => other == *c,
            Selector::Subset(subset) => subset.iter().any(|c| other == *c),
            Selector::All => true,
        }
    }
}

/// Represents the parameters of a single benchmark execution that collects a set of statistics.
pub trait TestCase: Debug + Clone + Hash + PartialEq + Eq + PartialOrd + Ord {
    /// A key type that identifies this test case ignoring the target field.
    /// Used for cross-target comparisons where we match test cases by everything except target.
    type KeyWithoutTarget: Debug + Clone + Hash + PartialEq + Eq;

    fn key_without_target(&self) -> Self::KeyWithoutTarget;
}

#[derive(Debug)]
pub struct SeriesResponse<Case, T> {
    pub test_case: Case,
    pub series: T,
}

impl<TestCase, T> SeriesResponse<TestCase, T> {
    pub fn map<U>(self, m: impl FnOnce(T) -> U) -> SeriesResponse<TestCase, U> {
        let SeriesResponse {
            test_case: key,
            series,
        } = self;
        SeriesResponse {
            test_case: key,
            series: m(series),
        }
    }

    pub fn interpolate(self) -> SeriesResponse<TestCase, Interpolate<T>>
    where
        T: Iterator,
        T::Item: Point,
    {
        self.map(|s| Interpolate::new(s))
    }
}

pub trait BenchmarkQuery: Debug + Clone {
    type TestCase: TestCase;

    #[allow(async_fn_in_trait)]
    async fn execute(
        &self,
        connection: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, StatisticSeries>>, String>;

    /// Execute the query and return all samples (not just the min value).
    /// Returns a nested structure: for each series, for each artifact, a vector of all sample values.
    #[allow(async_fn_in_trait)]
    async fn execute_all_samples(
        &self,
        connection: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, AllSamplesSeries>>, String>;

    /// Execute the query, fetching all samples and aggregating them with the given method.
    /// Returns `StatisticSeries` (single value per artifact) using the chosen aggregation.
    /// For `Aggregation::Min` this is equivalent to `execute()`.
    #[allow(async_fn_in_trait)]
    async fn execute_aggregated(
        &self,
        connection: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        aggregation: Aggregation,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, StatisticSeries>>, String> {
        if aggregation == Aggregation::Min {
            return self.execute(connection, index, artifact_ids).await;
        }
        let all = self
            .execute_all_samples(connection, index, artifact_ids)
            .await?;
        Ok(all
            .into_iter()
            .map(|sr| {
                let points: Vec<Option<f64>> = sr
                    .series
                    .samples
                    .collect::<Vec<_>>()
                    .into_iter()
                    .map(|samples| aggregate(&samples, aggregation))
                    .collect();
                let artifact_ids = sr.series.artifact_ids;
                SeriesResponse {
                    test_case: sr.test_case,
                    series: StatisticSeries {
                        artifact_ids,
                        points: points.into_iter(),
                    },
                }
            })
            .filter(|sr| sr.series.points.as_slice().iter().any(|v| v.is_some()))
            .collect())
    }
}

/// A series that contains all sample values for each artifact.
/// Unlike StatisticSeries which only contains the min value.
#[derive(Debug)]
pub struct AllSamplesSeries {
    pub artifact_ids: ArtifactIdIter,
    /// For each artifact, a vector of all sample values (or empty if no data)
    pub samples: std::vec::IntoIter<Vec<f64>>,
}

impl Iterator for AllSamplesSeries {
    type Item = (ArtifactId, Vec<f64>);
    fn next(&mut self) -> Option<Self::Item> {
        Some((self.artifact_ids.next()?, self.samples.next().unwrap()))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.artifact_ids.size_hint()
    }
}

// Compile benchmarks querying
#[derive(Clone, Hash, Eq, PartialEq, Debug)]
pub struct CompileBenchmarkQuery {
    benchmark: Selector<String>,
    scenario: Selector<Scenario>,
    profile: Selector<Profile>,
    backend: Selector<CodegenBackend>,
    metric: Selector<crate::Metric>,
    target: Selector<Target>,
}

impl CompileBenchmarkQuery {
    pub fn benchmark(mut self, selector: Selector<String>) -> Self {
        self.benchmark = selector;
        self
    }

    pub fn profile(mut self, selector: Selector<Profile>) -> Self {
        self.profile = selector;
        self
    }

    pub fn scenario(mut self, selector: Selector<Scenario>) -> Self {
        self.scenario = selector;
        self
    }

    pub fn backend(mut self, selector: Selector<CodegenBackend>) -> Self {
        self.backend = selector;
        self
    }

    pub fn target(mut self, selector: Selector<Target>) -> Self {
        self.target = selector;
        self
    }

    pub fn metric(mut self, selector: Selector<Metric>) -> Self {
        self.metric = selector.map(|v| v.as_str().into());
        self
    }

    /// Set metric filter using custom string selectors (for dynamic/custom metrics
    /// that may not be in the `Metric` enum).
    pub fn metric_custom(mut self, selector: Selector<String>) -> Self {
        self.metric = selector.map(|v| v.as_str().into());
        self
    }

    pub fn all_for_metric(metric: Metric) -> Self {
        Self {
            benchmark: Selector::All,
            profile: Selector::All,
            scenario: Selector::All,
            backend: Selector::All,
            metric: Selector::One(metric.as_str().into()),
            target: Selector::All,
        }
    }

    /// Like `all_for_metric` but accepts an arbitrary metric string,
    /// supporting custom metrics not in the `Metric` enum.
    pub fn all_for_metric_str(metric: &str) -> Self {
        Self {
            benchmark: Selector::All,
            profile: Selector::All,
            scenario: Selector::All,
            backend: Selector::All,
            metric: Selector::One(metric.into()),
            target: Selector::All,
        }
    }
}

impl Default for CompileBenchmarkQuery {
    fn default() -> Self {
        Self {
            benchmark: Selector::All,
            scenario: Selector::All,
            profile: Selector::All,
            backend: Selector::All,
            metric: Selector::All,
            target: Selector::All,
        }
    }
}

impl BenchmarkQuery for CompileBenchmarkQuery {
    type TestCase = CompileTestCase;

    async fn execute(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, StatisticSeries>>, String> {
        self.execute_with(conn, index, artifact_ids, false).await
    }

    async fn execute_all_samples(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, AllSamplesSeries>>, String> {
        self.execute_with(conn, index, artifact_ids, true).await
    }
}

impl CompileBenchmarkQuery {
    async fn execute_with<S>(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        all_samples: bool,
    ) -> Result<Vec<SeriesResponse<CompileTestCase, S>>, String>
    where
        S: FromCompileSeries,
    {
        let mut statistic_descriptions: Vec<_> = index
            .compile_statistic_descriptions()
            .filter(|(&(b, p, s, backend, target, metric), _)| {
                self.benchmark.matches(b)
                    && self.profile.matches(p)
                    && self.scenario.matches(s)
                    && self.backend.matches(backend)
                    && self.target.matches(target)
                    && self.metric.matches(metric)
            })
            .map(
                |(&(benchmark, profile, scenario, backend, target, metric), sid)| {
                    (
                        CompileTestCase {
                            benchmark,
                            profile,
                            scenario,
                            backend,
                            target,
                        },
                        metric,
                        sid,
                    )
                },
            )
            .collect();

        statistic_descriptions.sort_unstable();

        let sids: Vec<_> = statistic_descriptions
            .iter()
            .map(|(_, _, sid)| *sid)
            .collect();

        let aids = artifact_ids
            .iter()
            .map(|(aid, tag)| index.lookup_artifact_with_tag(aid, *tag))
            .collect::<Vec<_>>();

        let raw_data = if all_samples {
            S::from_all_samples(
                conn.get_compile_pstats_all_samples(&sids, &aids).await,
                artifact_ids.clone(),
                Aggregation::default(),
            )
        } else {
            S::from_min_values(
                conn.get_compile_pstats(&sids, &aids).await,
                artifact_ids.clone(),
            )
        };

        Ok(raw_data
            .into_iter()
            .zip(statistic_descriptions)
            .filter(|(series, _)| S::has_data(series))
            .map(|(series, (test_case, metric, _))| SeriesResponse {
                series: S::maybe_convert_units(series, metric.as_str()),
                test_case,
            })
            .collect())
    }
}

/// Trait for converting raw database results into series types.
/// Allows unified handling of both min-value and all-samples queries.
trait FromCompileSeries: Sized {
    /// Check if the series has any data
    fn has_data(series: &Self) -> bool;
    /// Convert from min values (existing behavior)
    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self>;
    /// Convert from all samples (new behavior)
    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        aggregation: Aggregation,
    ) -> Vec<Self>;
    /// Apply unit conversion if needed (e.g., ms to seconds for cpu-clock)
    fn maybe_convert_units(self, metric: &str) -> Self;
}

impl FromCompileSeries for StatisticSeries {
    fn has_data(series: &Self) -> bool {
        series.points.as_slice().iter().any(|v| v.is_some())
    }

    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|points| StatisticSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                points: points.into_iter(),
            })
            .collect()
    }

    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        aggregation: Aggregation,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|samples| StatisticSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                points: samples
                    .into_iter()
                    .map(|s| aggregate(&s, aggregation))
                    .collect::<Vec<_>>()
                    .into_iter(),
            })
            .collect()
    }

    fn maybe_convert_units(self, metric: &str) -> Self {
        if metric == "cpu-clock" || metric == "task-clock" {
            StatisticSeries {
                artifact_ids: self.artifact_ids,
                points: self
                    .points
                    .collect::<Vec<_>>()
                    .into_iter()
                    .map(|p| p.map(|v| v / 1000.0))
                    .collect::<Vec<_>>()
                    .into_iter(),
            }
        } else {
            self
        }
    }
}

impl FromCompileSeries for AllSamplesSeries {
    fn has_data(series: &Self) -> bool {
        series.samples.as_slice().iter().any(|v| !v.is_empty())
    }

    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|points| AllSamplesSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                samples: points
                    .into_iter()
                    .map(|p| p.into_iter().collect())
                    .collect::<Vec<_>>()
                    .into_iter(),
            })
            .collect()
    }

    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        _aggregation: Aggregation,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|samples| AllSamplesSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                samples: samples.into_iter(),
            })
            .collect()
    }

    fn maybe_convert_units(self, metric: &str) -> Self {
        if metric == "cpu-clock" || metric == "task-clock" {
            AllSamplesSeries {
                artifact_ids: self.artifact_ids,
                samples: self
                    .samples
                    .collect::<Vec<_>>()
                    .into_iter()
                    .map(|s| s.into_iter().map(|v| v / 1000.0).collect())
                    .collect::<Vec<_>>()
                    .into_iter(),
            }
        } else {
            self
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct CompileTestCase {
    pub benchmark: Benchmark,
    pub profile: Profile,
    pub scenario: Scenario,
    pub backend: CodegenBackend,
    pub target: Target,
}

/// A key for CompileTestCase that ignores the target field.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct CompileTestCaseKeyWithoutTarget {
    pub benchmark: Benchmark,
    pub profile: Profile,
    pub scenario: Scenario,
    pub backend: CodegenBackend,
}

impl TestCase for CompileTestCase {
    type KeyWithoutTarget = CompileTestCaseKeyWithoutTarget;

    fn key_without_target(&self) -> Self::KeyWithoutTarget {
        CompileTestCaseKeyWithoutTarget {
            benchmark: self.benchmark,
            profile: self.profile,
            scenario: self.scenario,
            backend: self.backend,
        }
    }
}

// Runtime benchmarks querying
#[derive(Clone, Hash, Eq, PartialEq, Debug)]
pub struct RuntimeBenchmarkQuery {
    benchmark: Selector<String>,
    target: Selector<Target>,
    metric: Selector<crate::Metric>,
}

impl RuntimeBenchmarkQuery {
    pub fn benchmark(mut self, selector: Selector<String>) -> Self {
        self.benchmark = selector;
        self
    }

    pub fn target(mut self, selector: Selector<Target>) -> Self {
        self.target = selector;
        self
    }

    pub fn metric(mut self, selector: Selector<Metric>) -> Self {
        self.metric = selector.map(|v| v.as_str().into());
        self
    }

    /// Set metric filter using custom string selectors (for dynamic/custom metrics
    /// that may not be in the `Metric` enum).
    pub fn metric_custom(mut self, selector: Selector<String>) -> Self {
        self.metric = selector.map(|v| v.as_str().into());
        self
    }

    pub fn all_for_metric(metric: Metric) -> Self {
        Self {
            benchmark: Selector::All,
            target: Selector::All,
            metric: Selector::One(metric.as_str().into()),
        }
    }

    /// Like `all_for_metric` but accepts an arbitrary metric string,
    /// supporting custom metrics not in the `Metric` enum.
    pub fn all_for_metric_str(metric: &str) -> Self {
        Self {
            benchmark: Selector::All,
            target: Selector::All,
            metric: Selector::One(metric.into()),
        }
    }
}

impl Default for RuntimeBenchmarkQuery {
    fn default() -> Self {
        Self {
            benchmark: Selector::All,
            target: Selector::All,
            metric: Selector::All,
        }
    }
}

impl BenchmarkQuery for RuntimeBenchmarkQuery {
    type TestCase = RuntimeTestCase;

    async fn execute(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, StatisticSeries>>, String> {
        self.execute_with(conn, index, artifact_ids, false).await
    }

    async fn execute_all_samples(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<Self::TestCase, AllSamplesSeries>>, String> {
        self.execute_with(conn, index, artifact_ids, true).await
    }
}

impl RuntimeBenchmarkQuery {
    async fn execute_with<S>(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        all_samples: bool,
    ) -> Result<Vec<SeriesResponse<RuntimeTestCase, S>>, String>
    where
        S: FromRuntimeSeries,
    {
        let mut statistic_descriptions: Vec<_> = index
            .runtime_statistic_descriptions()
            .filter(|(&(b, t, m), _)| {
                self.benchmark.matches(b) && self.target.matches(t) && self.metric.matches(m)
            })
            .map(|(&(benchmark, target, _), sid)| (RuntimeTestCase { benchmark, target }, sid))
            .collect();

        statistic_descriptions.sort_unstable();

        let sids: Vec<_> = statistic_descriptions.iter().map(|(_, sid)| *sid).collect();

        let aids = artifact_ids
            .iter()
            .map(|(aid, tag)| index.lookup_artifact_with_tag(aid, *tag))
            .collect::<Vec<_>>();

        let raw_data = if all_samples {
            S::from_all_samples(
                conn.get_runtime_pstats_all_samples(&sids, &aids).await,
                artifact_ids,
                Aggregation::default(),
            )
        } else {
            S::from_min_values(conn.get_runtime_pstats(&sids, &aids).await, artifact_ids)
        };

        Ok(raw_data
            .into_iter()
            .zip(statistic_descriptions)
            .filter(|(series, _)| S::has_data(series))
            .map(|(series, (test_case, _))| SeriesResponse { series, test_case })
            .collect())
    }
}

/// Trait for converting raw runtime database results into series types.
trait FromRuntimeSeries: Sized {
    fn has_data(series: &Self) -> bool;
    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self>;
    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        aggregation: Aggregation,
    ) -> Vec<Self>;
}

impl FromRuntimeSeries for StatisticSeries {
    fn has_data(series: &Self) -> bool {
        series.points.as_slice().iter().any(|v| v.is_some())
    }

    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|points| StatisticSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                points: points.into_iter(),
            })
            .collect()
    }

    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        aggregation: Aggregation,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|samples| StatisticSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                points: samples
                    .into_iter()
                    .map(|s| aggregate(&s, aggregation))
                    .collect::<Vec<_>>()
                    .into_iter(),
            })
            .collect()
    }
}

impl FromRuntimeSeries for AllSamplesSeries {
    fn has_data(series: &Self) -> bool {
        series.samples.as_slice().iter().any(|v| !v.is_empty())
    }

    fn from_min_values(
        data: Vec<Vec<Option<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|points| AllSamplesSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                samples: points
                    .into_iter()
                    .map(|p| p.into_iter().collect())
                    .collect::<Vec<_>>()
                    .into_iter(),
            })
            .collect()
    }

    fn from_all_samples(
        data: Vec<Vec<Vec<f64>>>,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
        _aggregation: Aggregation,
    ) -> Vec<Self> {
        data.into_iter()
            .map(|samples| AllSamplesSeries {
                artifact_ids: ArtifactIdIter::new(artifact_ids.clone()),
                samples: samples.into_iter(),
            })
            .collect()
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RuntimeTestCase {
    pub benchmark: Benchmark,
    pub target: Target,
}

impl TestCase for RuntimeTestCase {
    type KeyWithoutTarget = Benchmark;

    fn key_without_target(&self) -> Self::KeyWithoutTarget {
        self.benchmark
    }
}

/// A test case for a JSON metric in runtime benchmarks.
/// Includes the metric name since JSON metrics need to be distinguished.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct RuntimeJsonTestCase {
    pub benchmark: Benchmark,
    pub target: Target,
    pub metric: Metric,
}

/// A key for RuntimeJsonTestCase that ignores the target field.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct RuntimeJsonTestCaseKeyWithoutTarget {
    pub benchmark: Benchmark,
    pub metric: Metric,
}

impl TestCase for RuntimeJsonTestCase {
    type KeyWithoutTarget = RuntimeJsonTestCaseKeyWithoutTarget;

    fn key_without_target(&self) -> Self::KeyWithoutTarget {
        RuntimeJsonTestCaseKeyWithoutTarget {
            benchmark: self.benchmark,
            metric: self.metric.clone(),
        }
    }
}

/// Result of a JSON metric query for a single series across artifacts.
#[derive(Debug)]
pub struct JsonValueSeries {
    /// One JSON string per artifact (or None if no JSON data for that artifact)
    pub values: Vec<Option<String>>,
}

impl RuntimeBenchmarkQuery {
    /// Execute a query that returns JSON values instead of numeric values.
    /// Returns series that have json_value data for the given artifacts.
    pub async fn execute_json(
        &self,
        conn: &mut dyn Connection,
        index: &Index,
        artifact_ids: Arc<Vec<(ArtifactId, Tag)>>,
    ) -> Result<Vec<SeriesResponse<RuntimeJsonTestCase, JsonValueSeries>>, String> {
        let mut statistic_descriptions: Vec<_> = index
            .runtime_statistic_descriptions()
            .filter(|(&(b, t, m), _)| {
                self.benchmark.matches(b) && self.target.matches(t) && self.metric.matches(m)
            })
            .map(|(&(benchmark, target, metric), sid)| {
                (
                    RuntimeJsonTestCase {
                        benchmark,
                        target,
                        metric: Metric::from_str_inner(metric.as_str()),
                    },
                    sid,
                )
            })
            .collect();

        statistic_descriptions.sort_unstable();

        let sids: Vec<_> = statistic_descriptions.iter().map(|(_, sid)| *sid).collect();

        let aids = artifact_ids
            .iter()
            .map(|(aid, tag)| index.lookup_artifact_with_tag(aid, *tag))
            .collect::<Vec<_>>();

        let json_data = conn.get_runtime_json_pstats(&sids, &aids).await;

        Ok(json_data
            .into_iter()
            .zip(statistic_descriptions)
            .filter(|(values, _)| values.iter().any(|v| v.is_some()))
            .map(|(values, (test_case, _))| SeriesResponse {
                test_case,
                series: JsonValueSeries { values },
            })
            .collect())
    }
}