diff --git a/docs/deep-dive/deep-dive-clustering.md b/docs/deep-dive/deep-dive-clustering.md index e27543d..1b68f57 100644 --- a/docs/deep-dive/deep-dive-clustering.md +++ b/docs/deep-dive/deep-dive-clustering.md @@ -780,6 +780,7 @@ Multi-region fan-out rewrites SELECT into per-region partial aggregations, then | `first` / `last` | Carry bucket time; pick min/max time across regions | | `distinct`, `count(distinct)`, `stddev` | Set / moment partial merge | | `percentile`, `median`, `mode`, `spread` | **Error** — not supported across regions yet | +| `top`, `bottom`, `sample` | Each region returns its own N points per group; the coordinator ranks the union again (exact for `top`/`bottom`), then applies `LIMIT`/`OFFSET` | Global **LIMIT**, **OFFSET**, and **ORDER BY** apply on the coordinator after merging (per-region SQL fetches `limit+offset` rows). diff --git a/docs/user-guide/reference.md b/docs/user-guide/reference.md index 6786a9c..351670b 100644 --- a/docs/user-guide/reference.md +++ b/docs/user-guide/reference.md @@ -243,6 +243,22 @@ SELECT [, ...] FROM | `MODE(field)` | Most frequent value | | `DISTINCT(field)` | Unique values | +### Selector Functions + +These return up to N stored points per group (each `GROUP BY time()` bucket +and tag set), each at its own time, in time order. + +| Function | Description | +|----------|-------------| +| `TOP(field, N)` | The N largest values; the earliest point wins a tie | +| `TOP(field, tag[, tag...], N)` | The N largest values, at most one per tag-value combination; the tags are returned as columns | +| `BOTTOM(field, N)` / `BOTTOM(field, tag[, tag...], N)` | The N smallest values, likewise | +| `SAMPLE(field, N)` | N points picked at random | + +Other fields and tags may be selected beside them (`SELECT top("value", 3), "host"`), taken from the same points, but not other functions. `LIMIT` and `OFFSET` apply to the points returned. + +A raw field cannot be grouped by time: `SELECT "value" ... GROUP BY time(1m)` fails with `GROUP BY requires at least one aggregate function`, as in InfluxDB. + ### Transform Functions | Function | Description | diff --git a/hyperbytedb/src/application/query_service.rs b/hyperbytedb/src/application/query_service.rs index 4674426..4cedc71 100644 --- a/hyperbytedb/src/application/query_service.rs +++ b/hyperbytedb/src/application/query_service.rs @@ -2221,7 +2221,11 @@ async fn execute_select_from_source( select_with_expanded_group_by(select_stmt, &sub_tag_keys); if svc.shard_routing.is_some() { - let inner = execute_measurement_query( + // Times in nanoseconds, so the outer query gets a real `time` + // column back; it saw only the JSON's `__time` (a string or + // the client's epoch unit), and failed on every subquery + // that filters or buckets by time. + let mut inner = execute_measurement_query( svc, query_db, &rp, @@ -2229,26 +2233,27 @@ async fn execute_select_from_source( sub_stmt, time_min, time_max, - epoch, + Some("ns"), group_by_tags, ) .await?; if inner.is_empty() { return Ok(vec![]); } + stamp_aggregate_time(sub_stmt, &mut inner, Some("ns"), time_min); let inner_json = series_results_to_json_each_row(&inner)?; let escaped = escape_clickhouse_string_literal(&inner_json); - let sub_source_sql = format!("(SELECT * FROM format(JSONEachRow, '{escaped}'))"); + let sub_source_sql = format!( + "(SELECT * EXCEPT (`__time`), fromUnixTimestamp64Nano(toInt64(`__time`)) AS `time` FROM format(JSONEachRow, '{escaped}'))" + ); let outer_sql = to_clickhouse::translate_with_source(&select_stmt_expanded, &sub_source_sql)?; - return run_series_query( - svc, - &outer_sql, - sub_source, - epoch, - &resolved_group_by_tags, - ) - .await; + let series = + run_series_query(svc, &outer_sql, sub_source, epoch, &resolved_group_by_tags) + .await?; + return Ok(crate::domain::sharding::multi_selector::strip_hidden( + series, + )); } let sub_sql = to_clickhouse::translate_native_table( @@ -2258,15 +2263,21 @@ async fn execute_select_from_source( sub_series_join, Some((time_min, time_max)), )?; - // A GROUP BY time() inner query exposes its bucket column under the - // internal `__time` alias; as a FROM source it must expose `time` - // so the outer query's bucketing/ordering can reference it. - let sub_sql = to_clickhouse::rename_time_bucket_alias(&sub_sql); - let outer_sql = to_clickhouse::translate_with_source( - &select_stmt_expanded, - &format!("({sub_sql})"), + let sub_source_sql = to_clickhouse::subquery_source( + &sub_stmt_expanded, + &sub_sql, + time_min, + sub_effective_mapping.as_ref(), )?; - run_series_query(svc, &outer_sql, sub_source, epoch, &resolved_group_by_tags).await + let outer_sql = + to_clickhouse::translate_with_source(&select_stmt_expanded, &sub_source_sql)?; + let series = + run_series_query(svc, &outer_sql, sub_source, epoch, &resolved_group_by_tags) + .await?; + // An outer TOP/BOTTOM/SAMPLE's ranking columns. + Ok(crate::domain::sharding::multi_selector::strip_hidden( + series, + )) } } } @@ -2284,6 +2295,50 @@ async fn execute_measurement_query( epoch: Option<&str>, group_by_tags: &[String], ) -> Result, HyperbytedbError> { + // TOP/BOTTOM/SAMPLE rows carry hidden ranking columns; a sharded result + // is also ranked again over every region's rows. + if let Some(sel) = to_clickhouse::multi_selector(stmt)? { + let named = to_clickhouse::with_output_names(stmt); + if let Some(ctx) = &svc.shard_routing { + // `rerank` applies LIMIT and OFFSET, so nothing below may: the + // regions, and also the local table a measurement without a + // shard space falls back to, which applied them a second time. + let mut unlimited = stmt.clone(); + unlimited.limit = None; + unlimited.offset = None; + let series = execute_sharded_measurement_query( + svc, + ctx, + db, + rp, + measurement, + &unlimited, + time_min, + time_max, + epoch, + group_by_tags, + ) + .await?; + return Ok(crate::domain::sharding::multi_selector::rerank( + &sel, &named, series, + )); + } + let series = execute_local_measurement_query( + svc, + db, + rp, + measurement, + stmt, + time_min, + time_max, + epoch, + group_by_tags, + ) + .await?; + return Ok(crate::domain::sharding::multi_selector::strip_hidden( + series, + )); + } if let Some(ctx) = &svc.shard_routing { return execute_sharded_measurement_query( svc, @@ -2697,7 +2752,12 @@ async fn execute_sharded_measurement_query( &effective_stmt, ) }; - let is_global_aggregate = !is_materialized_dest && select_has_true_aggregate(&effective_stmt); + // TOP/BOTTOM/SAMPLE: each region returns its own N per group and the + // caller ranks the union again, then applies LIMIT and OFFSET (it passes + // the statement without them). + let multi_selector = to_clickhouse::multi_selector(&effective_stmt)?.is_some(); + let is_global_aggregate = + !is_materialized_dest && !multi_selector && select_has_true_aggregate(&effective_stmt); let selected_regions: Vec = region_selection .regions(space) .into_iter() @@ -2707,6 +2767,7 @@ async fn execute_sharded_measurement_query( metrics::histogram!("hyperbytedb_query_regions").record(selected_regions.len() as f64); stage.lap("route"); let sharded_plan = if !is_materialized_dest + && !multi_selector && (selected_regions.len() > 1 || (space.regions.len() > 1 && is_global_aggregate)) { let mut plan = prepare_sharded_region_query(&effective_stmt)?; @@ -2838,6 +2899,9 @@ async fn execute_sharded_measurement_query( if parts.is_empty() { return Ok(Vec::new()); } + if multi_selector && !is_materialized_dest { + return Ok(parts.into_iter().flat_map(first_statement_series).collect()); + } if parts.len() == 1 && !is_materialized_dest && sharded_plan.is_none() { let Some(part) = parts.pop() else { return Ok(Vec::new()); diff --git a/hyperbytedb/src/domain/sharding/mod.rs b/hyperbytedb/src/domain/sharding/mod.rs index 2c70741..5fb67f9 100644 --- a/hyperbytedb/src/domain/sharding/mod.rs +++ b/hyperbytedb/src/domain/sharding/mod.rs @@ -1,4 +1,5 @@ pub mod location_cache; +pub mod multi_selector; pub mod ops; pub mod placement; pub mod query_merge; diff --git a/hyperbytedb/src/domain/sharding/multi_selector.rs b/hyperbytedb/src/domain/sharding/multi_selector.rs new file mode 100644 index 0000000..6a134d8 --- /dev/null +++ b/hyperbytedb/src/domain/sharding/multi_selector.rs @@ -0,0 +1,270 @@ +//! Finishing TOP, BOTTOM and SAMPLE results. +//! +//! Each region of a sharded measurement returns its own N points per group, +//! so the coordinator ranks the union again ([`rerank`]). The top N of a +//! union of per-region top-N lists is the global top N, also with tag +//! arguments: a tag's best point overall is its best point in some region, +//! and is in that region's list unless N other tags beat it there, and so +//! overall. SAMPLE picks N at random from the regions' samples. +//! +//! Every path then drops the hidden ranking columns ([`strip_hidden`]). + +use std::cmp::Ordering; +use std::collections::HashMap; +use std::hash::BuildHasher; + +use indexmap::IndexMap; +use serde_json::Value; + +use crate::domain::query_result::SeriesResult; +use crate::timeseriesql::ast::SelectStatement; +use crate::timeseriesql::to_clickhouse::{BUCKET_COL, MultiSelector, SelectorKind, TIME_NS_COL}; + +/// Remove the hidden ranking columns. +#[must_use] +pub fn strip_hidden(mut series: Vec) -> Vec { + for s in &mut series { + let keep: Vec = (0..s.columns.len()) + .filter(|&i| s.columns[i] != BUCKET_COL && s.columns[i] != TIME_NS_COL) + .collect(); + if keep.len() == s.columns.len() { + continue; + } + s.columns = keep.iter().map(|&i| s.columns[i].clone()).collect(); + for row in &mut s.values { + *row = keep + .iter() + .map(|&i| row.get(i).cloned().unwrap_or(Value::Null)) + .collect(); + } + } + series +} + +/// Rank the regions' rows again: at most `sel.n` per bucket and series, in +/// time order, then the statement's LIMIT and OFFSET per series. The hidden +/// columns are dropped. +#[must_use] +pub fn rerank( + sel: &MultiSelector, + stmt: &SelectStatement, + series: Vec, +) -> Vec { + // The same series from several regions becomes one. + let mut merged: IndexMap<(String, Vec<(String, String)>), SeriesResult> = IndexMap::new(); + for s in series { + let mut key_tags: Vec<(String, String)> = s + .tags + .as_ref() + .map(|t| t.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) + .unwrap_or_default(); + key_tags.sort(); + match merged.entry((s.name.clone(), key_tags)) { + indexmap::map::Entry::Vacant(e) => { + e.insert(s); + } + indexmap::map::Entry::Occupied(mut e) => { + let into = e.get_mut(); + let at: Vec> = into + .columns + .iter() + .map(|c| s.columns.iter().position(|x| x == c)) + .collect(); + for row in s.values { + into.values.push( + at.iter() + .map(|i| i.and_then(|i| row.get(i).cloned()).unwrap_or(Value::Null)) + .collect(), + ); + } + } + } + } + + let desc = stmt.order_by.as_ref().is_some_and(|o| o.time_desc); + let random = std::collections::hash_map::RandomState::new(); + let mut out = Vec::with_capacity(merged.len()); + for (_, mut s) in merged { + let col = |name: &str| s.columns.iter().position(|c| c == name); + let (Some(bucket), Some(tns), Some(value)) = + (col(BUCKET_COL), col(TIME_NS_COL), col(&sel.output)) + else { + out.push(s); + continue; + }; + let tag_cols: Vec = sel.tags.iter().filter_map(|t| col(t)).collect(); + let num = |row: &[Value], i: usize| row.get(i).and_then(Value::as_f64); + let int = |row: &[Value], i: usize| row.get(i).and_then(Value::as_i64).unwrap_or(0); + + let mut groups: IndexMap>> = IndexMap::new(); + for row in std::mem::take(&mut s.values) { + groups.entry(int(&row, bucket)).or_default().push(row); + } + let mut kept = Vec::new(); + for (_, mut rows) in groups { + match sel.kind { + SelectorKind::Top | SelectorKind::Bottom => { + rows.sort_by(|a, b| { + let by_value = match (num(a, value), num(b, value)) { + (Some(x), Some(y)) => x.total_cmp(&y), + (Some(_), None) => Ordering::Greater, + (None, Some(_)) => Ordering::Less, + (None, None) => Ordering::Equal, + }; + let by_value = if sel.kind == SelectorKind::Top { + by_value.reverse() + } else { + by_value + }; + by_value.then_with(|| int(a, tns).cmp(&int(b, tns))) + }); + if !tag_cols.is_empty() { + let mut seen = HashMap::new(); + rows.retain(|r| { + let key: Vec = tag_cols + .iter() + .map(|&i| r.get(i).cloned().unwrap_or(Value::Null)) + .collect(); + seen.insert(serde_json::to_string(&key).unwrap_or_default(), ()) + .is_none() + }); + } + } + SelectorKind::Sample => { + rows.sort_by_key(|r| random.hash_one(int(r, tns))); + } + } + rows.truncate(usize::try_from(sel.n).unwrap_or(usize::MAX)); + kept.extend(rows); + } + kept.sort_by(|a, b| { + let o = int(a, tns).cmp(&int(b, tns)); + if desc { o.reverse() } else { o } + }); + let offset = usize::try_from(stmt.offset.unwrap_or(0)).unwrap_or(usize::MAX); + let limit = stmt + .limit + .map_or(usize::MAX, |l| usize::try_from(l).unwrap_or(usize::MAX)); + s.values = kept.into_iter().skip(offset).take(limit).collect(); + if !s.values.is_empty() { + out.push(s); + } + } + strip_hidden(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn part(tags: Option<&str>, rows: &[(i64, i64, f64, &str)]) -> SeriesResult { + SeriesResult { + name: "cpu".into(), + tags: tags.map(|h| HashMap::from([("dc".to_string(), h.to_string())])), + columns: vec![ + "time".into(), + TIME_NS_COL.into(), + BUCKET_COL.into(), + "top".into(), + "host".into(), + ], + values: rows + .iter() + .map(|(t, b, v, h)| vec![json!(t), json!(t), json!(b), json!(v), json!(h)]) + .collect(), + partial: None, + } + } + + fn sel(kind: SelectorKind, tags: &[&str], n: u64) -> MultiSelector { + MultiSelector { + kind, + output: "top".into(), + field: "v".into(), + tags: tags.iter().map(|t| t.to_string()).collect(), + n, + } + } + + fn stmt() -> SelectStatement { + match crate::timeseriesql::parser::parse_query("SELECT top(v, 2) FROM cpu") + .unwrap() + .pop() + { + Some(crate::timeseriesql::ast::Statement::Select(s)) => s, + other => panic!("{other:?}"), + } + } + + /// Two regions' top 2 per bucket: the coordinator keeps the global top 2 + /// per bucket, earliest first on ties, in time order, and drops the + /// hidden columns. + #[test] + fn keeps_the_global_top_n_per_bucket() { + let r1 = part( + None, + &[(1, 0, 5.0, "a"), (2, 0, 9.0, "a"), (11, 10, 1.0, "a")], + ); + let r2 = part( + None, + &[(3, 0, 9.0, "b"), (4, 0, 7.0, "b"), (12, 10, 3.0, "b")], + ); + let out = rerank(&sel(SelectorKind::Top, &[], 2), &stmt(), vec![r1, r2]); + assert_eq!(out.len(), 1); + assert_eq!(out[0].columns, ["time", "top", "host"]); + let got: Vec<(i64, f64)> = out[0] + .values + .iter() + .map(|r| (r[0].as_i64().unwrap(), r[1].as_f64().unwrap())) + .collect(); + assert_eq!(got, [(2, 9.0), (3, 9.0), (11, 1.0), (12, 3.0)]); + } + + /// With a tag argument at most one point per tag value: host a's best + /// (9) and host b's best (8), not a's two best. + #[test] + fn a_tag_argument_keeps_one_point_per_value() { + let r1 = part(None, &[(1, 0, 9.0, "a"), (2, 0, 8.5, "a")]); + let r2 = part(None, &[(3, 0, 8.0, "b"), (4, 0, 1.0, "c")]); + let out = rerank(&sel(SelectorKind::Top, &["host"], 2), &stmt(), vec![r1, r2]); + let hosts: Vec<&str> = out[0] + .values + .iter() + .map(|r| r[2].as_str().unwrap()) + .collect(); + assert_eq!(hosts, ["a", "b"]); + } + + #[test] + fn bottom_keeps_the_smallest_and_series_stay_apart() { + let east = part(Some("east"), &[(1, 0, 5.0, "a"), (2, 0, 1.0, "a")]); + let west = part(Some("west"), &[(3, 0, 4.0, "b"), (4, 0, 2.0, "b")]); + let east2 = part(Some("east"), &[(5, 0, 3.0, "c")]); + let out = rerank( + &sel(SelectorKind::Bottom, &[], 1), + &stmt(), + vec![east, west, east2], + ); + assert_eq!(out.len(), 2); + assert_eq!(out[0].values[0][1], json!(1.0)); + assert_eq!(out[1].values[0][1], json!(2.0)); + } + + #[test] + fn sample_keeps_n_per_bucket() { + let r1 = part( + None, + &[(1, 0, 1.0, "a"), (2, 0, 2.0, "a"), (3, 0, 3.0, "a")], + ); + let r2 = part(None, &[(4, 0, 4.0, "b"), (11, 10, 5.0, "b")]); + let out = rerank(&sel(SelectorKind::Sample, &[], 2), &stmt(), vec![r1, r2]); + let buckets: Vec = out[0] + .values + .iter() + .map(|r| r[0].as_i64().unwrap()) + .collect(); + assert_eq!(buckets.len(), 3, "two from bucket 0, one from bucket 10"); + assert!(buckets.windows(2).all(|w| w[0] < w[1]), "time order"); + } +} diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/mod.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/mod.rs index 809b221..46bd164 100644 --- a/hyperbytedb/src/timeseriesql/to_clickhouse/mod.rs +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/mod.rs @@ -8,12 +8,14 @@ mod conditions; mod materialized_view; mod rename; mod select; +mod selectors; mod time_bounds; #[cfg(test)] mod tests; pub use rename::rename_time_bucket_alias; +pub use selectors::{BUCKET_COL, MultiSelector, SelectorKind, TIME_NS_COL, multi_selector}; pub use time_bounds::extract_time_bounds; pub use coalesce::{build_coalesced_fact_view, build_coalesced_fact_view_with_row_meta}; @@ -26,7 +28,7 @@ pub use materialized_view::{ translate_materialized_view_series_select, translate_select_into, translate_select_into_native, }; pub use select::{ - legacy_field_alias, select_has_true_aggregate, select_output_field_name, + legacy_field_alias, select_has_true_aggregate, select_output_field_name, subquery_source, translate_native_table, translate_native_table_unordered, translate_with_source, with_output_names, }; diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs index 51b9274..41124df 100644 --- a/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/select.rs @@ -62,6 +62,15 @@ fn translate_inner_ordered( // so `mean(a), mean(b)` is `mean`, `mean_1` in the SQL and the merge alike. let named = with_output_names(stmt); let stmt = &named; + if let Some(sel) = super::selectors::multi_selector(stmt)? { + return super::selectors::translate_multi_selector( + stmt, + &sel, + from_source, + mapping, + series, + ); + } let mut out = String::new(); // InfluxQL treats a GROUP BY time() query without an explicit fill() as @@ -129,6 +138,13 @@ fn translate_inner_ordered( .fields .iter() .any(|f| matches!(f.expr, Expr::Star | Expr::Wildcard)); + // As InfluxDB: raw points cannot be bucketed. ClickHouse would reject the + // SQL with NOT_AN_AGGREGATE, which surfaced as a chDB error. + if has_group_by_time && !has_aggregate { + return Err(HyperbytedbError::QueryParse( + "GROUP BY requires at least one aggregate function".to_string(), + )); + } // True aggregates collapse rows; bare window transforms (difference("v"), // moving_average("v", n), ...) stay per-point and must keep the raw `time` // column and per-point ordering like raw selects. @@ -392,6 +408,47 @@ pub fn translate_with_source( ) -> Result { translate_inner(stmt, source, None, None, None) } + +/// `sub_sql` (the translation of `sub`) as an outer query's FROM source, +/// with a `time` column as InfluxDB gives a subquery's points: the bucket +/// for GROUP BY time(), the selected point's time for a lone selector, else +/// the WHERE clause's lower bound (`time_min`, epoch 0 without one). +/// +/// Renaming the `__time` alias to `time` inside `sub_sql` made ClickHouse +/// read `time` in its GROUP BY as the bucket, not the column, so every +/// subquery over an aggregate failed (NOT_AN_AGGREGATE, or no `time`). +pub fn subquery_source( + sub: &SelectStatement, + sub_sql: &str, + time_min: Option, + mapping: Option<&ColumnMapping>, +) -> Result { + let named = with_output_names(sub); + let has_group_by_time = named + .group_by + .as_ref() + .and_then(|gb| gb.time_dimension()) + .is_some(); + let has_star = named + .fields + .iter() + .any(|f| matches!(f.expr, Expr::Star | Expr::Wildcard)); + // TOP/BOTTOM/SAMPLE keep their points' own `time`. + if super::selectors::multi_selector(&named)?.is_some() { + return Ok(format!("({sub_sql})")); + } + if has_group_by_time || lone_selector_time_sql(&named, mapping)?.is_some() { + return Ok(format!( + "(SELECT * EXCEPT (`__time`), `__time` AS `time` FROM ({sub_sql}))" + )); + } + if select_has_true_aggregate(&named) && !has_star { + let stamp = nanos_to_ch_timestamp(time_min.unwrap_or(0)); + return Ok(format!("(SELECT *, {stamp} AS `time` FROM ({sub_sql}))")); + } + // Raw points (and per-point transforms) carry their own `time`. + Ok(format!("({sub_sql})")) +} pub(super) fn expr_references_tag(expr: &Expr, m: &ColumnMapping) -> bool { match expr { Expr::Identifier(name) => m.tag_keys.contains(name), diff --git a/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs b/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs new file mode 100644 index 0000000..3889c4c --- /dev/null +++ b/hyperbytedb/src/timeseriesql/to_clickhouse/selectors.rs @@ -0,0 +1,305 @@ +//! TOP, BOTTOM and SAMPLE: selectors that return up to N stored points per +//! group (GROUP BY time() bucket and tags), each at its own time, as InfluxDB. +//! +//! The SQL ranks points in a window per group and keeps the first N. It also +//! returns two hidden columns, [`BUCKET_COL`] and [`TIME_NS_COL`], so a +//! sharded query's coordinator can rank the regions' rows again +//! ([`crate::domain::sharding::multi_selector`]); they are removed before the +//! result is returned. + +use std::fmt::Write; + +use crate::domain::column_mapping::ColumnMapping; +use crate::error::HyperbytedbError; +use crate::timeseriesql::ast::*; + +use super::SeriesJoin; +use super::aggregates::translate_aggregate_arg; +use super::conditions::{quote_identifier, translate_expr}; +use super::select::{ + build_from_source, select_output_field_name, select_tag_column_sql, time_bucket_expr, + translate_field, with_output_names, +}; + +/// Each row's group bucket, in nanoseconds (0 without GROUP BY time()). +pub const BUCKET_COL: &str = "__bucket"; +/// Each row's point time, in nanoseconds. +pub const TIME_NS_COL: &str = "__tns"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SelectorKind { + Top, + Bottom, + Sample, +} + +/// A statement's TOP, BOTTOM or SAMPLE call, parsed. +#[derive(Debug, Clone)] +pub struct MultiSelector { + pub kind: SelectorKind, + /// Output column of the selected values (`top` unless aliased). + pub output: String, + /// The field ranked. + pub field: String, + /// TOP/BOTTOM's tag arguments: at most one point per combination. + pub tags: Vec, + pub n: u64, +} + +fn kind_of(name: &str) -> Option { + match name.to_ascii_uppercase().as_str() { + "TOP" => Some(SelectorKind::Top), + "BOTTOM" => Some(SelectorKind::Bottom), + "SAMPLE" => Some(SelectorKind::Sample), + _ => None, + } +} + +fn contains_multi_selector(expr: &Expr) -> bool { + match expr { + Expr::Call(fc) => { + kind_of(&fc.name).is_some() || fc.args.iter().any(contains_multi_selector) + } + Expr::BinaryExpr(be) => { + contains_multi_selector(&be.left) || contains_multi_selector(&be.right) + } + Expr::UnaryExpr(_, e) => contains_multi_selector(e), + _ => false, + } +} + +fn ident(expr: &Expr) -> Option<&str> { + match expr { + Expr::Identifier(n) | Expr::FieldRef { name: n, .. } => Some(n), + _ => None, + } +} + +/// The statement's TOP, BOTTOM or SAMPLE, or None without one. Errors as +/// InfluxDB does when it is combined with other functions or misused. +pub fn multi_selector(stmt: &SelectStatement) -> Result, HyperbytedbError> { + if !stmt.fields.iter().any(|f| contains_multi_selector(&f.expr)) { + return Ok(None); + } + let named = with_output_names(stmt); + let mut found = None; + for field in &named.fields { + match &field.expr { + Expr::Call(fc) if kind_of(&fc.name).is_some() && found.is_none() => { + found = Some((fc, select_output_field_name(field))); + } + Expr::Identifier(_) | Expr::FieldRef { .. } => {} + Expr::Call(fc) if kind_of(&fc.name).is_some() => { + return Err(combined_error(&fc.name)); + } + _ => { + let name = named + .fields + .iter() + .find_map(|f| match &f.expr { + Expr::Call(fc) if kind_of(&fc.name).is_some() => Some(fc.name.clone()), + _ => None, + }) + .unwrap_or_else(|| "top".to_string()); + return Err(combined_error(&name)); + } + } + } + let Some((fc, output)) = found else { + // Nested inside another expression, e.g. `top(v, 2) * 2`. + return Err(HyperbytedbError::QueryParse( + "top(), bottom() and sample() cannot be used inside an expression".to_string(), + )); + }; + let kind = kind_of(&fc.name).unwrap_or(SelectorKind::Top); + let lname = fc.name.to_ascii_lowercase(); + let min_args = 2; + if fc.args.len() < min_args || (kind == SelectorKind::Sample && fc.args.len() != 2) { + return Err(HyperbytedbError::QueryParse(format!( + "invalid number of arguments for {lname}, expected {}, got {}", + if kind == SelectorKind::Sample { + "2".to_string() + } else { + "at least 2".to_string() + }, + fc.args.len() + ))); + } + let field = ident(&fc.args[0]).ok_or_else(|| { + HyperbytedbError::QueryParse(format!( + "expected first argument to be a field in {lname}(), found {:?}", + fc.args[0] + )) + })?; + let n = match fc.args.last() { + Some(Expr::IntegerLiteral(n)) if *n >= 1 => *n as u64, + Some(Expr::IntegerLiteral(n)) => { + return Err(HyperbytedbError::QueryParse(format!( + "limit ({n}) in {lname} function must be at least 1" + ))); + } + other => { + return Err(HyperbytedbError::QueryParse(format!( + "expected integer as last argument in {lname}(), found {other:?}" + ))); + } + }; + let tags = fc.args[1..fc.args.len() - 1] + .iter() + .map(|a| { + ident(a).map(str::to_string).ok_or_else(|| { + HyperbytedbError::QueryParse(format!( + "only fields or tags are allowed in {lname}(), found {a:?}" + )) + }) + }) + .collect::, _>>()?; + Ok(Some(MultiSelector { + kind, + output: output.unwrap_or(lname), + field: field.to_string(), + tags, + n, + })) +} + +fn combined_error(name: &str) -> HyperbytedbError { + HyperbytedbError::QueryParse(format!( + "selector function {}() cannot be combined with other functions", + name.to_ascii_lowercase() + )) +} + +/// The SQL for a statement with a TOP, BOTTOM or SAMPLE call `sel`. +pub(super) fn translate_multi_selector( + stmt: &SelectStatement, + sel: &MultiSelector, + from_source: &str, + mapping: Option<&ColumnMapping>, + series: Option>, +) -> Result { + let named = with_output_names(stmt); + let stmt = &named; + let tz = stmt.timezone.as_deref(); + let gb_tags: Vec<&str> = stmt + .group_by + .as_ref() + .map(|gb| gb.tag_dimensions()) + .unwrap_or_default(); + let bucket = match stmt.group_by.as_ref().and_then(|gb| gb.time_dimension()) { + Some(Dimension::Time { interval, offset }) => format!( + "toUnixTimestamp64Nano(toDateTime64({}, 9))", + time_bucket_expr(interval, offset.as_ref(), tz) + ), + _ => "toInt64(0)".to_string(), + }; + let value = translate_aggregate_arg(&Expr::Identifier(sel.field.clone()), mapping)?; + let out = quote_identifier(&sel.output)?; + let time = "`time`"; + + // Points, with the value under the selector's name and every column the + // result needs. + let mut cols = vec![ + time.to_string(), + format!("toUnixTimestamp64Nano(toDateTime64({time}, 9)) AS `{TIME_NS_COL}`"), + format!("{bucket} AS `{BUCKET_COL}`"), + format!("{value} AS {out}"), + ]; + let mut seen: Vec = vec![sel.output.clone()]; + for tag in gb_tags + .iter() + .copied() + .chain(sel.tags.iter().map(String::as_str)) + { + if !seen.iter().any(|s| s == tag) { + cols.push(select_tag_column_sql(tag, mapping)?); + seen.push(tag.to_string()); + } + } + for field in &stmt.fields { + if let Expr::Identifier(n) | Expr::FieldRef { name: n, .. } = &field.expr { + let name = field.alias.clone().unwrap_or_else(|| n.clone()); + if n != "time" && !seen.contains(&name) { + cols.push(translate_field( + field, + false, + 0.0, + stmt.group_by.as_ref(), + mapping, + )?); + seen.push(name); + } + } + } + // Laid out as the other translations, so the region and tombstone + // predicates spliced in later (by line: `\nWHERE `, and into the inner + // query of a `SELECT * FROM (\n...\n) ` wrapper) filter the points + // before they are ranked. + let source = build_from_source(from_source, series, mapping, stmt); + let mut points = format!("SELECT {}\nFROM {}", cols.join(", "), source.from); + write!(points, "\nWHERE {value} IS NOT NULL")?; + if let Some(cond) = source.outer_condition.as_ref() { + points.push_str(" AND ("); + translate_expr(cond, &mut points, true, mapping)?; + points.push(')'); + } + + let group: Vec = std::iter::once(format!("`{BUCKET_COL}`")) + .chain( + gb_tags + .iter() + .map(|t| quote_identifier(t)) + .collect::, _>>()?, + ) + .collect(); + let order = match sel.kind { + SelectorKind::Top => format!("{out} DESC, {time} ASC"), + SelectorKind::Bottom => format!("{out} ASC, {time} ASC"), + SelectorKind::Sample => "rand64()".to_string(), + }; + // TOP/BOTTOM's tag arguments: each combination's best point first. + if !sel.tags.is_empty() { + let mut per_tag = group.clone(); + for t in &sel.tags { + let q = quote_identifier(t)?; + if !per_tag.contains(&q) { + per_tag.push(q); + } + } + points = format!( + "SELECT * FROM (\n{points}\n) QUALIFY row_number() OVER (PARTITION BY {} ORDER BY {order}) = 1", + per_tag.join(", ") + ); + } + let mut sql = format!( + "SELECT * FROM (\n{points}\n) QUALIFY row_number() OVER (PARTITION BY {} ORDER BY {order}) <= {}", + group.join(", "), + sel.n + ); + + // As a raw select: series by tag, each in time order. On the `) ` line: + // the splicers look for clauses only inside the wrapper. + let desc = stmt.order_by.as_ref().is_some_and(|o| o.time_desc); + let mut order_by: Vec = gb_tags + .iter() + .map(|t| Ok(format!("{} ASC", quote_identifier(t)?))) + .collect::>()?; + order_by.push(format!("{time} {}", if desc { "DESC" } else { "ASC" })); + write!(sql, " ORDER BY {}", order_by.join(", "))?; + if let Some(limit) = stmt.limit { + let offset = stmt.offset.unwrap_or(0); + if gb_tags.is_empty() { + write!(sql, " LIMIT {offset}, {limit}")?; + } else { + let by = gb_tags + .iter() + .map(|t| quote_identifier(t)) + .collect::, _>>()? + .join(", "); + write!(sql, " LIMIT {offset}, {limit} BY ({by})")?; + } + } else if let Some(offset) = stmt.offset { + write!(sql, " OFFSET {offset}")?; + } + Ok(sql) +} diff --git a/hyperbytedb/tests/compat/query_tests.rs b/hyperbytedb/tests/compat/query_tests.rs index 1e4b2de..0835cfa 100644 --- a/hyperbytedb/tests/compat/query_tests.rs +++ b/hyperbytedb/tests/compat/query_tests.rs @@ -1171,3 +1171,266 @@ async fn columns_and_time_follow_influxql() { let (cols, _) = shape("SELECT v + w FROM cpu").await; assert_eq!(cols, ["time", "v_w"]); } + +// --------------------------------------------------------------------------- +// Subqueries over aggregates; TOP, BOTTOM and SAMPLE +// --------------------------------------------------------------------------- + +/// 60 s of points for hosts a and b: `usage` = base + (7s mod 23) + 0.5, +/// base 0 for a and 100 for b, so b's maximum, 122.5, is at s = 13, 36 and +/// 59, and each host's minimum at s = 0, 23 and 46. Starts at +/// 2001-09-09T01:46:40Z (10^9 s). +async fn two_hosts() -> TestContext { + let ctx = TestContext::new().unwrap(); + ctx.metadata.create_database("testdb").await.unwrap(); + let mut lines = Vec::new(); + for (host, base) in [("a", 0), ("b", 100)] { + for s in 0..60i64 { + lines.push(format!( + "cpu,host={host} usage={}.5,n={s}i {}", + base + (s * 7) % 23, + (1_000_000_000 + s) * 1_000_000_000 + )); + } + } + ctx.write_and_flush("testdb", &lines.join("\n")) + .await + .unwrap(); + ctx +} + +const RANGE: &str = "time >= 1000000000s AND time < 1000000060s"; + +/// The statement's series, failing on any error. +async fn series_of( + ctx: &TestContext, + q: &str, +) -> Vec { + let resp = ctx + .query("testdb", q) + .await + .unwrap_or_else(|e| panic!("{q}: {e}")); + assert!( + resp.results[0].error.is_none(), + "{q}: {:?}", + resp.results[0].error + ); + resp.results[0].series.clone().unwrap_or_default() +} + +/// The error a statement fails with, as a client sees it. +async fn error_of(ctx: &TestContext, q: &str) -> String { + match ctx.query("testdb", q).await { + Err(e) => e.to_string(), + Ok(resp) => resp.results[0] + .error + .clone() + .unwrap_or_else(|| panic!("{q} should fail")), + } +} + +fn rows(s: &hyperbytedb::domain::query_result::SeriesResult) -> Vec> { + s.values.clone() +} + +/// (time, value) for a series whose first two columns are those; JSON `112` +/// and `112.0` compare unequal as values. +fn points(s: &hyperbytedb::domain::query_result::SeriesResult) -> Vec<(String, f64)> { + s.values + .iter() + .map(|r| (r[0].as_str().unwrap().to_string(), r[1].as_f64().unwrap())) + .collect() +} + +/// A subquery over a GROUP BY time() aggregate: renaming its bucket to +/// `time` inside its own SQL made ClickHouse read `time` in its GROUP BY as +/// the bucket, and every such query failed with NOT_AN_AGGREGATE. +#[tokio::test] +#[serial(chdb)] +async fn subquery_over_time_buckets() { + let ctx = two_hosts().await; + let q = format!( + "SELECT max(m) FROM (SELECT mean(usage) AS m FROM cpu WHERE {RANGE} GROUP BY time(10s), host) WHERE {RANGE} GROUP BY time(30s)" + ); + let series = series_of(&ctx, &q).await; + assert_eq!(series[0].columns, ["time", "max"]); + // Host b's 10 s means are 111.3, 112.3 | 111, 112, 110.7 | 111.7; the + // 30 s buckets start at :30, :00 and :30. + assert_eq!( + points(&series[0]), + [ + ("2001-09-09T01:46:30Z".to_string(), 112.3), + ("2001-09-09T01:47:00Z".to_string(), 112.0), + ("2001-09-09T01:47:30Z".to_string(), 111.7), + ] + ); +} + +/// A subquery over an aggregate without buckets had no `time` column at all. +/// Its points take the query's lower time bound, as in InfluxDB. +#[tokio::test] +#[serial(chdb)] +async fn subquery_over_an_aggregate_without_buckets() { + let ctx = two_hosts().await; + let q = format!( + "SELECT max(m) FROM (SELECT mean(usage) AS m FROM cpu WHERE {RANGE} GROUP BY host) WHERE {RANGE}" + ); + let series = series_of(&ctx, &q).await; + assert_eq!( + rows(&series[0]), + vec![vec![ + serde_json::json!("2001-09-09T01:46:40Z"), + serde_json::json!(111.5) + ]] + ); +} + +#[tokio::test] +#[serial(chdb)] +async fn a_raw_field_cannot_be_grouped_by_time() { + let ctx = two_hosts().await; + for q in [ + format!("SELECT usage FROM cpu WHERE {RANGE} GROUP BY time(20s)"), + format!("SELECT * FROM cpu WHERE {RANGE} GROUP BY time(20s)"), + ] { + let err = error_of(&ctx, &q).await; + assert!( + err.contains("GROUP BY requires at least one aggregate function"), + "{q}: {err}" + ); + } +} + +/// TOP keeps the largest points at their own times, earliest first on ties, +/// in time order. +#[tokio::test] +#[serial(chdb)] +async fn top_returns_the_largest_points() { + let ctx = two_hosts().await; + let series = series_of( + &ctx, + &format!("SELECT top(usage, 2) FROM cpu WHERE {RANGE}"), + ) + .await; + assert_eq!(series[0].columns, ["time", "top"]); + assert_eq!( + rows(&series[0]), + vec![ + vec![ + serde_json::json!("2001-09-09T01:46:53Z"), + serde_json::json!(122.5) + ], + vec![ + serde_json::json!("2001-09-09T01:47:16Z"), + serde_json::json!(122.5) + ], + ] + ); + + // With a tag argument, one point per host; with another field beside + // it, that field from the same point. + let series = series_of( + &ctx, + &format!("SELECT top(usage, host, 2), n FROM cpu WHERE {RANGE}"), + ) + .await; + assert_eq!(series[0].columns, ["time", "top", "n", "host"]); + let got: Vec<(f64, i64, String)> = series[0] + .values + .iter() + .map(|r| { + ( + r[1].as_f64().unwrap(), + r[2].as_i64().unwrap(), + r[3].as_str().unwrap().to_string(), + ) + }) + .collect(); + assert_eq!(got, [(122.5, 13, "b".into()), (22.5, 13, "a".into())]); +} + +/// Per GROUP BY time() bucket and per series. +#[tokio::test] +#[serial(chdb)] +async fn bottom_and_sample_per_bucket() { + let ctx = two_hosts().await; + let series = series_of( + &ctx, + &format!("SELECT bottom(usage, 1) FROM cpu WHERE {RANGE} GROUP BY time(20s), host"), + ) + .await; + assert_eq!(series.len(), 2); + for (s, min) in series.iter().zip([0.5, 100.5]) { + let times: Vec<&str> = s.values.iter().map(|r| r[0].as_str().unwrap()).collect(); + assert_eq!( + times, + [ + "2001-09-09T01:46:40Z", + "2001-09-09T01:47:03Z", + "2001-09-09T01:47:26Z" + ], + "{s:?}" + ); + assert!(s.values.iter().all(|r| r[1].as_f64() == Some(min)), "{s:?}"); + } + + // 30 s buckets from :30: 20 + 30 + 10 points, two picked from each. + let series = series_of( + &ctx, + &format!("SELECT sample(usage, 2) FROM cpu WHERE {RANGE} GROUP BY time(30s)"), + ) + .await; + assert_eq!(series[0].columns, ["time", "sample"]); + assert_eq!(series[0].values.len(), 6); +} + +#[tokio::test] +#[serial(chdb)] +async fn top_cannot_be_combined_with_other_functions() { + let ctx = two_hosts().await; + let err = error_of( + &ctx, + &format!("SELECT top(usage, 2), mean(usage) FROM cpu WHERE {RANGE}"), + ) + .await; + assert!( + err.contains("selector function top() cannot be combined"), + "{err}" + ); + let err = error_of( + &ctx, + &format!("SELECT top(usage, 0) FROM cpu WHERE {RANGE}"), + ) + .await; + assert!(err.contains("must be at least 1"), "{err}"); +} + +/// TOP over a subquery, and an aggregate over a TOP subquery. +#[tokio::test] +#[serial(chdb)] +async fn top_and_subqueries() { + let ctx = two_hosts().await; + let series = series_of( + &ctx, + &format!( + "SELECT top(m, 1) FROM (SELECT mean(usage) AS m FROM cpu WHERE {RANGE} GROUP BY time(10s)) WHERE {RANGE}" + ), + ) + .await; + assert_eq!(series[0].columns, ["time", "top"], "no ranking columns"); + assert_eq!( + rows(&series[0]), + vec![vec![ + serde_json::json!("2001-09-09T01:46:50Z"), + serde_json::json!(62.3) + ]] + ); + let series = series_of( + &ctx, + &format!( + "SELECT mean(top) FROM (SELECT top(usage, 3) FROM cpu WHERE {RANGE}) WHERE {RANGE}" + ), + ) + .await; + assert_eq!(series[0].values[0][1], serde_json::json!(122.5)); +} diff --git a/hyperbytedb/tests/sharding_cluster_integration.rs b/hyperbytedb/tests/sharding_cluster_integration.rs index 3a658d3..1ce5580 100644 --- a/hyperbytedb/tests/sharding_cluster_integration.rs +++ b/hyperbytedb/tests/sharding_cluster_integration.rs @@ -638,6 +638,242 @@ async fn multi_region_group_by_time_merges_buckets() { } } +/// TOP, BOTTOM and SAMPLE across two regions: each region returns its own +/// N per group and the coordinator ranks the union again, so the result is +/// the global N, not N per region. +#[tokio::test] +#[serial(chdb)] +async fn multi_region_top_bottom_and_sample() { + let dir = tempfile::tempdir().unwrap(); + let nodes = + start_sharded_three_node_cluster(dir.path(), ShardedClusterOptions::default()).await; + bootstrap_two_regions(&nodes, "sharddb", "autogen", "cpu").await; + let client = reqwest::Client::new(); + for node in &nodes { + create_db(&client, &node.url, "sharddb").await; + } + let (low, high) = hosts_for_region_split("cpu", u64::MAX / 2); + let s = 1_000_000_000u64; + // Low region: 10 @0s, 20 @30s, 25 @50s. High region: 30 @10s, 40 @70s, + // 5 @80s. + let body = format!( + "cpu,host={low} value=10 0\n\ + cpu,host={low} value=20 {}\n\ + cpu,host={low} value=25 {}\n\ + cpu,host={high} value=30 {}\n\ + cpu,host={high} value=40 {}\n\ + cpu,host={high} value=5 {}", + 30 * s, + 50 * s, + 10 * s, + 70 * s, + 80 * s + ); + let resp = write_line(&client, &nodes[0].url, "sharddb", &body).await; + assert!(resp.status().is_success(), "write: {}", resp.status()); + for node in &nodes { + flush_node(node).await; + } + let range = format!("time >= 0 AND time < {}", 120 * s); + let values = |resp: &QueryResponse| -> Vec> { + let result = &resp.results[0]; + assert!(result.error.is_none(), "{:?}", result.error); + result + .series + .as_ref() + .expect("series") + .iter() + .map(|series| { + assert!( + !series.columns.iter().any(|c| c.starts_with("__")), + "{:?}", + series.columns + ); + series + .values + .iter() + .map(|r| r[1].as_f64().unwrap()) + .collect() + }) + .collect() + }; + for node in &nodes { + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!("SELECT top(value, 2) FROM cpu WHERE {range}"), + ) + .await; + assert_eq!( + values(&resp), + [vec![30.0, 40.0]], + "{}: global top 2, time order", + node.url + ); + + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!("SELECT top(value, 1) FROM cpu WHERE {range} GROUP BY time(1m)"), + ) + .await; + assert_eq!( + values(&resp), + [vec![30.0, 40.0]], + "{}: per minute", + node.url + ); + + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!("SELECT bottom(value, 1) FROM cpu WHERE {range} GROUP BY host"), + ) + .await; + let mut per_host = values(&resp); + per_host.sort_by(|a, b| a[0].total_cmp(&b[0])); + assert_eq!(per_host, [vec![5.0], vec![10.0]], "{}: per host", node.url); + + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!("SELECT top(value, 3) FROM cpu WHERE {range} LIMIT 1 OFFSET 1"), + ) + .await; + assert_eq!( + values(&resp), + [vec![25.0]], + "{}: LIMIT/OFFSET after ranking", + node.url + ); + + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!("SELECT sample(value, 2) FROM cpu WHERE {range}"), + ) + .await; + assert_eq!( + values(&resp)[0].len(), + 2, + "{}: two points overall", + node.url + ); + } +} + +/// A sharded node queries a measurement it has no shard space for (here the +/// space is dropped from the map, the rows stay) from its own table. That +/// local SQL applied LIMIT and OFFSET, and the coordinator's re-rank applied +/// them again: `top(v, 3) LIMIT 1 OFFSET 1` returned nothing. +#[tokio::test] +#[serial(chdb)] +async fn top_limit_offset_on_the_local_fallback_applies_once() { + let dir = tempfile::tempdir().unwrap(); + let node = start_sharded_single_node(dir.path(), ShardedClusterOptions::default()).await; + let client = reqwest::Client::new(); + create_db(&client, &node.url, "fdb").await; + let s = 1_000_000_000u64; + let body = format!( + "cpu,host=a value=10 {}\ncpu,host=a value=30 {}\ncpu,host=a value=20 {}", + s, + 2 * s, + 3 * s + ); + let resp = write_line(&client, &node.url, "fdb", &body).await; + assert!(resp.status().is_success(), "write: {}", resp.status()); + flush_node(&node).await; + node.shard_map + .apply_op(ShardMapOp::DropSpace { + db: "fdb".into(), + rp: Some("autogen".into()), + measurement: Some("cpu".into()), + }) + .await + .unwrap(); + let snap = node.shard_map.snapshot().await.unwrap(); + node.location_cache.refresh_from_map(&snap); + + let resp = query_parsed( + &client, + &node.url, + "fdb", + "SELECT top(value, 3) FROM cpu WHERE time >= 0 AND time < 10s LIMIT 1 OFFSET 1", + ) + .await; + let result = &resp.results[0]; + assert!(result.error.is_none(), "{:?}", result.error); + let series = result.series.as_ref().expect("one point, not none"); + let values: Vec = series[0] + .values + .iter() + .map(|r| r[1].as_f64().unwrap()) + .collect(); + assert_eq!( + values, + [30.0], + "top 3 in time order is 10, 30, 20; OFFSET 1 LIMIT 1 is 30" + ); +} + +/// A subquery over time buckets on a sharded measurement: the inner result +/// is gathered from the regions and the outer query runs over it. +#[tokio::test] +#[serial(chdb)] +async fn multi_region_subquery_over_time_buckets() { + let dir = tempfile::tempdir().unwrap(); + let nodes = + start_sharded_three_node_cluster(dir.path(), ShardedClusterOptions::default()).await; + bootstrap_two_regions(&nodes, "sharddb", "autogen", "cpu").await; + let client = reqwest::Client::new(); + for node in &nodes { + create_db(&client, &node.url, "sharddb").await; + } + let (low, high) = hosts_for_region_split("cpu", u64::MAX / 2); + let s = 1_000_000_000u64; + let body = format!( + "cpu,host={low} value=10 0\n\ + cpu,host={low} value=20 {}\n\ + cpu,host={high} value=30 {}\n\ + cpu,host={high} value=40 {}", + 30 * s, + 70 * s, + 100 * s + ); + let resp = write_line(&client, &nodes[0].url, "sharddb", &body).await; + assert!(resp.status().is_success(), "write: {}", resp.status()); + for node in &nodes { + flush_node(node).await; + } + let range = format!("time >= 0 AND time < {}", 120 * s); + for node in &nodes { + let resp = query_parsed( + &client, + &node.url, + "sharddb", + &format!( + "SELECT max(m) FROM (SELECT mean(value) AS m FROM cpu WHERE {range} GROUP BY time(1m), host) WHERE {range} GROUP BY time(2m)" + ), + ) + .await; + let result = &resp.results[0]; + assert!(result.error.is_none(), "{}: {:?}", node.url, result.error); + let series = result.series.as_ref().expect("series"); + let maxes: Vec = series[0] + .values + .iter() + .map(|r| r[1].as_f64().unwrap()) + .collect(); + // Per minute and host: 15 (low, :00), 35 (high, :01): one 2 m bucket. + assert_eq!(maxes, [35.0], "{}", node.url); + } +} + /// Across regions, a lone selector reports the winning point's time and a /// plain aggregate epoch 0, as on one node. The winners sit in /// different regions: max and last in one, min and first in the other.