Merge pull request #944 from rylev/simplify-query-interface

rylev · web-flow · commit 6a5dfd02b44e · 2021-07-30T10:46:25.000+02:00
Simplify query interface
diff --git a/site/src/comparison.rs b/site/src/comparison.rs
@@ -247,7 +247,7 @@ async fn compare_given_commits(
 
     // `responses` contains series iterators. The first element in the iterator is the data
     // for `a` and the second is the data for `b`
-    let mut responses = ctxt.query::<Option<f64>>(query.clone(), aids).await?;
+    let mut responses = ctxt.statistic_series(query.clone(), aids).await?;
 
     let conn = ctxt.conn().await;
     Ok(Some(Comparison {
@@ -492,7 +492,7 @@ impl BenchmarkVariances {
             master_commits,
         ));
         let mut previous_commit_series = ctxt
-            .query::<Option<f64>>(query, previous_commits.clone())
+            .statistic_series(query, previous_commits.clone())
             .await?;
 
         let mut variance_data: HashMap<String, BenchmarkVariance> = HashMap::new();
diff --git a/site/src/selector.rs b/site/src/selector.rs
@@ -25,7 +25,6 @@ use crate::db::{ArtifactId, Profile, Scenario};
 use crate::interpolate::Interpolate;
 use crate::load::SiteCtxt;
 
-use async_trait::async_trait;
 use collector::Bound;
 use database::{Benchmark, Commit, Index, Lookup, Metric, QueryLabel};
 
@@ -268,13 +267,6 @@ impl<T> SeriesResponse<T> {
     }
 }
 
-pub trait Series: Sized
-where
-    Self: Iterator<Item = (ArtifactId, <Self as Series>::Element)>,
-{
-    type Element: Sized;
-}
-
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
 pub struct Path {
     path: Vec<PathComponent>,
@@ -381,46 +373,6 @@ impl Query {
     }
 }
 
-#[async_trait]
-pub trait SeriesElement: Sized {
-    async fn query<'a>(
-        ctxt: &'a SiteCtxt,
-        artifact_ids: Arc<Vec<ArtifactId>>,
-        query: Query,
-    ) -> Result<Vec<SeriesResponse<Box<dyn Iterator<Item = (ArtifactId, Self)> + Send + 'a>>>, String>;
-}
-
-fn handle_results<'a, E>(
-    results: Vec<
-        Result<Vec<SeriesResponse<Box<dyn Iterator<Item = (ArtifactId, E)> + Send + 'a>>>, String>,
-    >,
-) -> Result<Vec<SeriesResponse<Box<dyn Iterator<Item = (ArtifactId, E)> + Send + 'a>>>, String> {
-    let mut ok = None;
-    let mut errs = Vec::new();
-    for res in results {
-        match (res, ok.is_some()) {
-            (Ok(r), false) => {
-                ok = Some(r);
-            }
-            (Ok(_), true) => panic!("two series successfully expanded"),
-            (Err(e), _) => errs.push(e),
-        }
-    }
-
-    ok.ok_or_else(|| {
-        format!(
-            "Failed to process query; fix one of these errors: {}",
-            errs.into_iter().fold(String::new(), |mut acc, err| {
-                if !acc.is_empty() {
-                    acc.push_str("; or ");
-                }
-                acc.push_str(&err);
-                acc
-            })
-        )
-    })
-}
-
 #[derive(Debug, Clone)]
 pub struct SelfProfileData {
     pub query_data: Vec<QueryData>,
@@ -454,88 +406,29 @@ impl QueryData {
     }
 }
 
-#[async_trait]
-impl SeriesElement for Option<SelfProfileData> {
-    async fn query<'a>(
-        ctxt: &'a SiteCtxt,
-        artifact_ids: Arc<Vec<ArtifactId>>,
+impl SiteCtxt {
+    pub async fn statistic_series(
+        &self,
         query: Query,
-    ) -> Result<
-        Vec<
-            SeriesResponse<
-                Box<dyn Iterator<Item = (ArtifactId, Option<SelfProfileData>)> + Send + 'a>,
-            >,
-        >,
-        String,
-    > {
-        let results = vec![SelfProfile::expand_query(artifact_ids, ctxt, query.clone())
-            .await
-            .map(|sr| {
-                sr.into_iter()
-                    .map(|sr| {
-                        sr.map(|r| {
-                            Box::new(r)
-                                as Box<
-                                    dyn Iterator<Item = (ArtifactId, Option<SelfProfileData>)>
-                                        + Send,
-                                >
-                        })
-                    })
-                    .collect()
-            })];
-        handle_results(results)
-    }
-}
-
-#[async_trait]
-impl SeriesElement for Option<f64> {
-    async fn query<'a>(
-        ctxt: &'a SiteCtxt,
         artifact_ids: Arc<Vec<ArtifactId>>,
-        query: Query,
-    ) -> Result<
-        Vec<SeriesResponse<Box<dyn Iterator<Item = (ArtifactId, Option<f64>)> + Send + 'a>>>,
-        String,
-    > {
-        let results = vec![
-            StatisticSeries::expand_query(artifact_ids.clone(), ctxt, query.clone())
-                .await
-                .map(|sr| {
-                    sr.into_iter()
-                        .map(|sr| {
-                            sr.map(|r| {
-                                Box::new(r)
-                                    as Box<dyn Iterator<Item = (ArtifactId, Option<f64>)> + Send>
-                            })
-                        })
-                        .collect()
-                }),
-            SelfProfileQueryTime::expand_query(artifact_ids.clone(), ctxt, query.clone())
-                .await
-                .map(|sr| {
-                    sr.into_iter()
-                        .map(|sr| {
-                            sr.map(|r| {
-                                Box::new(r)
-                                    as Box<dyn Iterator<Item = (ArtifactId, Option<f64>)> + Send>
-                            })
-                        })
-                        .collect()
-                }),
-        ];
+    ) -> Result<Vec<SeriesResponse<StatisticSeries>>, String> {
+        StatisticSeries::execute_query(artifact_ids.clone(), self, query.clone()).await
+    }
 
-        handle_results(results)
+    pub async fn self_profile(
+        &self,
+        query: Query,
+        artifact_ids: Arc<Vec<ArtifactId>>,
+    ) -> Result<Vec<SeriesResponse<SelfProfile>>, String> {
+        SelfProfile::execute_query(artifact_ids, self, query.clone()).await
     }
-}
 
-impl SiteCtxt {
-    pub async fn query<'a, E: SeriesElement>(
-        &'a self,
+    pub async fn self_profile_query_time(
+        &self,
         query: Query,
         artifact_ids: Arc<Vec<ArtifactId>>,
-    ) -> Result<Vec<SeriesResponse<Box<dyn Iterator<Item = (ArtifactId, E)> + Send + 'a>>>, String>
-    {
-        E::query(self, artifact_ids, query).await
+    ) -> Result<Vec<SeriesResponse<SelfProfileQueryTime>>, String> {
+        SelfProfileQueryTime::execute_query(artifact_ids, self, query.clone()).await
     }
 }
 
@@ -544,12 +437,8 @@ pub struct StatisticSeries {
     points: std::vec::IntoIter<Option<f64>>,
 }
 
-impl Series for StatisticSeries {
-    type Element = Option<f64>;
-}
-
 impl StatisticSeries {
-    async fn expand_query(
+    async fn execute_query(
         artifact_ids: Arc<Vec<ArtifactId>>,
         ctxt: &SiteCtxt,
         mut query: Query,
@@ -710,25 +599,8 @@ impl SelfProfile {
             points: res.into_iter(),
         }
     }
-}
-
-impl Iterator for SelfProfile {
-    type Item = (ArtifactId, Option<SelfProfileData>);
-    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()
-    }
-}
-
-impl Series for SelfProfile {
-    type Element = Option<SelfProfileData>;
-}
 
-impl SelfProfile {
-    async fn expand_query(
+    async fn execute_query(
         artifact_ids: Arc<Vec<ArtifactId>>,
         ctxt: &SiteCtxt,
         mut query: Query,
@@ -765,6 +637,17 @@ impl SelfProfile {
     }
 }
 
+impl Iterator for SelfProfile {
+    type Item = (ArtifactId, Option<SelfProfileData>);
+    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 struct SelfProfileQueryTime {
     artifact_ids: ArtifactIdIter,
     points: std::vec::IntoIter<Option<f64>>,
@@ -802,25 +685,8 @@ impl SelfProfileQueryTime {
             points: res.into_iter(),
         }
     }
-}
-
-impl Iterator for SelfProfileQueryTime {
-    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()
-    }
-}
-
-impl Series for SelfProfileQueryTime {
-    type Element = Option<f64>;
-}
-
-impl SelfProfileQueryTime {
-    async fn expand_query(
+    async fn execute_query(
         artifact_ids: Arc<Vec<ArtifactId>>,
         ctxt: &SiteCtxt,
         mut query: Query,
@@ -834,35 +700,35 @@ impl SelfProfileQueryTime {
         let index = ctxt.index.load();
         let mut series = index
             .all_query_series()
-            .filter(|tup| {
-                benchmark.matches(tup.0)
-                    && profile.matches(tup.1)
-                    && scenario.matches(tup.2)
-                    && ql.matches(tup.3)
+            .filter(|&&(b, p, s, q)| {
+                benchmark.matches(b) && profile.matches(p) && scenario.matches(s) && ql.matches(q)
             })
             .collect::<Vec<_>>();
 
         series.sort_unstable();
 
         let mut res = Vec::with_capacity(series.len());
-        for path in series {
+        for &(b, p, s, q) in series {
             res.push(SeriesResponse {
-                series: SelfProfileQueryTime::new(
-                    artifact_ids.clone(),
-                    ctxt,
-                    path.0,
-                    path.1,
-                    path.2,
-                    path.3,
-                )
-                .await,
+                series: SelfProfileQueryTime::new(artifact_ids.clone(), ctxt, b, p, s, q).await,
                 path: Path::new()
-                    .set(PathComponent::Benchmark(path.0))
-                    .set(PathComponent::Profile(path.1))
-                    .set(PathComponent::Scenario(path.2))
-                    .set(PathComponent::QueryLabel(path.3)),
+                    .set(PathComponent::Benchmark(b))
+                    .set(PathComponent::Profile(p))
+                    .set(PathComponent::Scenario(s))
+                    .set(PathComponent::QueryLabel(q)),
             });
         }
         Ok(res)
     }
 }
+
+impl Iterator for SelfProfileQueryTime {
+    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()
+    }
+}
diff --git a/site/src/server.rs b/site/src/server.rs
@@ -181,7 +181,7 @@ pub async fn handle_dashboard(ctxt: Arc<SiteCtxt>) -> ServerResult<dashboard::Re
             let mut cases = dashboard::Cases::default();
             for scenario in summary_scenarios.iter() {
                 let responses = ctxt
-                    .query::<Option<f64>>(
+                    .statistic_series(
                         query
                             .clone()
                             .set(Tag::Profile, selector::Selector::One(profile))
@@ -478,7 +478,7 @@ pub async fn handle_graph(
     let metric_selector = selector::Selector::One(body.stat.clone());
 
     let series = ctxt
-        .query::<Option<f64>>(
+        .statistic_series(
             selector::Query::new()
                 .set::<String>(selector::Tag::Benchmark, selector::Selector::All)
                 .set::<String>(selector::Tag::Profile, selector::Selector::All)
@@ -543,7 +543,7 @@ pub async fn handle_graph(
             std::collections::hash_map::Entry::Occupied(o) => *o.get(),
             std::collections::hash_map::Entry::Vacant(v) => {
                 let value = db::average(
-                    ctxt.query::<Option<f64>>(q, c.clone())
+                    ctxt.statistic_series(q, c.clone())
                         .await?
                         .into_iter()
                         .map(|sr| sr.interpolate().series)
@@ -555,7 +555,7 @@ pub async fn handle_graph(
             }
         };
         let averaged = db::average(
-            ctxt.query::<Option<f64>>(query.clone(), commits.clone())
+            ctxt.statistic_series(query.clone(), commits.clone())
                 .await?
                 .into_iter()
                 .map(|sr| sr.interpolate().series)
@@ -1078,9 +1078,7 @@ pub async fn handle_self_profile(
     }
 
     let commits = Arc::new(commits);
-    let mut sp_responses = ctxt
-        .query::<Option<selector::SelfProfileData>>(query.clone(), commits.clone())
-        .await?;
+    let mut sp_responses = ctxt.self_profile(query.clone(), commits.clone()).await?;
 
     if sp_responses.is_empty() {
         return Err(format!("no results found for {:?} in {:?}", query, commits));
@@ -1098,7 +1096,7 @@ pub async fn handle_self_profile(
     let mut sp_response = sp_responses.remove(0).series;
 
     let mut cpu_responses = ctxt
-        .query::<Option<f64>>(
+        .self_profile_query_time(
             query.clone().set(
                 Tag::Metric,
                 selector::Selector::One("cpu-clock".to_string()),
