diff --git a/crates/integration-tests/tests/sql_group_bound.rs b/crates/integration-tests/tests/sql_group_bound.rs index 4f6b65442..e3b2adce4 100644 --- a/crates/integration-tests/tests/sql_group_bound.rs +++ b/crates/integration-tests/tests/sql_group_bound.rs @@ -194,3 +194,30 @@ fn an_undeclared_sql_grouping_is_priced_at_the_groups_of_rows_or_the_rows_read() priced_groups_and_scan_rows(workload(DISTINCT_USERS, "relative", every_column())); assert_eq!(groups, std::collections::BTreeSet::from([625])); } + +/// Each endpoint is in one service: declaring `(service, endpoint)`'s 625 +/// combinations prices `GROUP BY service, endpoint` at 625 groups, where +/// the per-column product (25 x 625) is capped only by the 10,000 groups of +/// rows. +#[test] +fn a_declared_column_set_prices_a_sql_grouping_at_its_combinations() { + let hierarchical = + serde_json::json!({"region": 4, "service": 25, "endpoint": 625, "status": 4}); + let (groups, _) = + priced_groups_and_scan_rows(workload(DISTINCT_USERS, "relative", hierarchical.clone())); + assert_eq!(groups, std::collections::BTreeSet::from([10_000])); + let mut declared = workload(DISTINCT_USERS, "relative", hierarchical); + declared + .data_workload + .as_mut() + .unwrap() + .sources + .get_mut("http") + .unwrap() + .distinct_values_per_label_set = vec![asap_types::workload::LabelSetDistinctValues { + labels: vec!["service".to_string(), "endpoint".to_string()], + distinct_values: 625, + }]; + let (groups, _) = priced_groups_and_scan_rows(declared); + assert_eq!(groups, std::collections::BTreeSet::from([625])); +} diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index ff119a392..6a2be077f 100644 --- a/crates/plan-selection/src/lib.rs +++ b/crates/plan-selection/src/lib.rs @@ -1917,14 +1917,15 @@ fn values_per_result( /// An upper bound on the groups `node` returns, from the distinct values /// the sources it reads declare for its group keys -/// (`SourceFacts::distinct_values_per_label`): the product over the keys of -/// each key's declared count ([`declared_distinct_values`]). The grouping is +/// (`SourceFacts::distinct_values_per_label` and +/// `distinct_values_per_label_set`): the smallest product of declared counts +/// covering the keys ([`declared_distinct_combinations`]). The grouping is /// found below operators that only keep, drop or reorder rows, and through /// a summary's estimate (not a top-k, which returns several rows per /// group), merge and finalization. `None` when no such grouping is found, /// it is a `without (...)`, or a key is not a source column passed through -/// unchanged with a declared count: the declared counts are upper bounds, -/// never defaults. +/// unchanged covered by declared counts: the declared counts are upper +/// bounds, never defaults. fn declared_group_bound(node: &OperatorNode, data: &DataWorkload) -> Option { let grouped = |reduction: &Reduction, input: &OperatorNode| -> Option { let Reduction::Reduce(keys) = reduction else { @@ -1933,9 +1934,7 @@ fn declared_group_bound(node: &OperatorNode, data: &DataWorkload) -> Option if keys.is_without() { return None; } - keys.keys().iter().try_fold(1u64, |product, &key| { - Some(product.saturating_mul(declared_distinct_values(input, key, data)?)) - }) + declared_distinct_combinations(input, keys.keys(), data) }; match &node.operator { Operator::NonASAP(NonASAPOp::Aggregate { @@ -1988,34 +1987,41 @@ fn declared_group_bound(node: &OperatorNode, data: &DataWorkload) -> Option } } -/// The distinct values column `column` of `node` takes at most, when it is -/// a source column passed through unchanged: traced down through -/// operators that keep their input's columns (filter, sort, limit, time -/// range and shift), a projection that selects it, a grouped aggregate's -/// key, a per-series reduction's other columns, and each branch of a -/// union, to the scans, and summed over them from the label (or column) -/// counts their sources declare. `None` when it is computed or relabeled -/// on the way (its values are no longer the source's), or a source does -/// not declare it. -fn declared_distinct_values( +/// The distinct combinations of values columns `columns` of `node` take +/// together at most, when each is a source column passed through +/// unchanged: traced down through operators that keep their input's +/// columns (filter, sort, limit, time range and shift), a projection that +/// selects them, a grouped aggregate's keys, a per-series reduction's other +/// columns, and each branch of a union (whose combinations add), to the +/// scans, where the source's declared label and label-set counts bound them +/// ([`SourceFacts::declared_distinct_combinations`]). A join is not traced: +/// it can pair values of different rows. `None` when a column is computed +/// or relabeled on the way (its values are no longer the source's), or a +/// source's declared counts do not cover the columns. +/// +/// [`SourceFacts::declared_distinct_combinations`]: asap_types::workload::SourceFacts::declared_distinct_combinations +fn declared_distinct_combinations( node: &OperatorNode, - column: usize, + columns: &[usize], data: &DataWorkload, ) -> Option { match node.non_asap()? { NonASAPOp::Scan { source, .. } => { - let label = &node.schema.fields.get(column)?.name; + let labels = columns + .iter() + .map(|&column| Some(node.schema.fields.get(column)?.name.as_str())) + .collect::>>()?; data.sources .get(source_name(source))? - .distinct_values_per_label - .get(label) - .copied() + .declared_distinct_combinations(&labels) } NonASAPOp::Filter { child, .. } | NonASAPOp::Sort { child, .. } | NonASAPOp::Limit { child, .. } | NonASAPOp::TimeRange { child, .. } - | NonASAPOp::TimeShift { child, .. } => declared_distinct_values(child, column, data), + | NonASAPOp::TimeShift { child, .. } => { + declared_distinct_combinations(child, columns, data) + } // A grouped aggregate returns its keys first, unchanged; but a // `count_values` label named like a key takes that key's place // with the stringified sample values. @@ -2029,7 +2035,11 @@ fn declared_distinct_values( .iter() .any(|m| matches!(m, AggIntent::CountValues { .. })) => { - declared_distinct_values(child, *keys.keys().get(column)?, data) + let inputs = columns + .iter() + .map(|&column| keys.keys().get(column).copied()) + .collect::>>()?; + declared_distinct_combinations(child, &inputs, data) } // A per-series reduction keeps every column but the one it reduces. NonASAPOp::Aggregate { @@ -2046,25 +2056,31 @@ fn declared_distinct_values( ) .ok()?, }; - (column != reduced) - .then(|| declared_distinct_values(child, column, data)) + (!columns.contains(&reduced)) + .then(|| declared_distinct_combinations(child, columns, data)) .flatten() } - NonASAPOp::Project { cols, child, .. } => match cols.get(column)?.expr { - ScalarExpr::Column(input) => declared_distinct_values(child, input, data), - _ => None, - }, + NonASAPOp::Project { cols, child, .. } => { + let inputs = columns + .iter() + .map(|&column| match cols.get(column)?.expr { + ScalarExpr::Column(input) => Some(input), + _ => None, + }) + .collect::>>()?; + declared_distinct_combinations(child, &inputs, data) + } NonASAPOp::SetOp { kind: RelationalSetOpKind::Union, left, right, .. } => Some( - declared_distinct_values(left, column, data)? - .saturating_add(declared_distinct_values(right, column, data)?), + declared_distinct_combinations(left, columns, data)? + .saturating_add(declared_distinct_combinations(right, columns, data)?), ), NonASAPOp::Concat { children, .. } => children.iter().try_fold(0u64, |total, child| { - Some(total.saturating_add(declared_distinct_values(child, column, data)?)) + Some(total.saturating_add(declared_distinct_combinations(child, columns, data)?)) }), _ => None, } @@ -2452,7 +2468,8 @@ fn guarantee_check( "q{query}: the output's guarantee holds for each value, and the number \ of values one evaluation returns is not bound (no declared series \ count for a source a PromQL vector reads, and no declared \ - `distinct_values_per_label` for every group key, or a key that is not \ + `distinct_values_per_label` for every group key nor a \ + `distinct_values_per_label_set` covering them, or a key that is not \ an unchanged source column), so it does not meet \ whole-result target {target:?}, and \ the deployment has no evidence that it meets it (a target on the groups \ @@ -2890,13 +2907,141 @@ fn source_rows_per_ms(sources: &BTreeSet<&str>, data: &DataWorkload) -> Option( + node: &'a asap_types::ir::physical_export::PhysicalASAPDAGNode, + producers: &[PhysicalASAPNodeId], + nodes: &NodeMap<'a>, + unchanged_columns_of: &HashMap>, +) -> BTreeSet<&'a str> { + let output_names = || node.output_schema.fields.iter().map(|f| f.name.as_str()); + let first = || { + producers.first().map(|producer| { + ( + &unchanged_columns_of[producer], + &nodes[producer].output_schema, + ) + }) + }; + // The output columns passing input column `input[i]` on as itself. + let passed_on = |inputs: &mut dyn Iterator>| -> BTreeSet<&'a str> { + let Some((unchanged, schema)) = first() else { + return BTreeSet::new(); + }; + output_names() + .zip(inputs) + .filter(|(name, input)| { + input + .and_then(|input| schema.fields.get(input)) + .is_some_and(|field| field.name == *name && unchanged.contains(*name)) + }) + .map(|(name, _)| name) + .collect() + }; + let kept = || -> BTreeSet<&'a str> { + first().map_or_else(BTreeSet::new, |(unchanged, _)| { + output_names() + .filter(|name| unchanged.contains(name)) + .collect() + }) + }; + match &node.payload { + Payload::NonASAP(operator) => match operator { + NonASAPOp::Scan { .. } => output_names().collect(), + NonASAPOp::Filter { .. } + | NonASAPOp::Sort { .. } + | NonASAPOp::Limit { .. } + | NonASAPOp::Dedup { .. } + | NonASAPOp::TimeRange { .. } + | NonASAPOp::TimeShift { .. } => kept(), + NonASAPOp::Project { cols, .. } => { + passed_on(&mut cols.iter().map(|col| match col.expr { + ScalarExpr::Column(input) => Some(input), + _ => None, + })) + } + // A `count_values` label named like a key replaces it. + NonASAPOp::Aggregate { + reduction: Reduction::Reduce(keys), + measures, + .. + } if !keys.is_without() + && !measures + .iter() + .any(|m| matches!(m, AggIntent::CountValues { .. })) => + { + passed_on(&mut keys.keys().iter().map(|&key| Some(key))) + } + // A per-series reduction keeps every column but the ones it + // reduces and the ones it names. + NonASAPOp::Aggregate { + reduction: Reduction::PerEntity, + measures, + output_names: measure_names, + .. + } => { + let Some((_, schema)) = first() else { + return BTreeSet::new(); + }; + let mut columns = kept(); + for measure in measures { + for input in measure.input_cols() { + if let Some(field) = schema.fields.get(input) { + columns.remove(field.name.as_str()); + } + } + } + for name in measure_names { + columns.remove(name.as_str()); + } + columns + } + NonASAPOp::SetOp { + kind: RelationalSetOpKind::Union, + .. + } + | NonASAPOp::Concat { .. } => { + let mut columns = kept(); + for producer in producers { + columns.retain(|name| unchanged_columns_of[producer].contains(name)); + } + columns + } + _ => BTreeSet::new(), + }, + Payload::ASAP( + ASAPOp::SummaryAgg { .. } + | ASAPOp::PaneChain { .. } + | ASAPOp::SummaryMerge { .. } + | ASAPOp::SummaryEstimate { .. } + | ASAPOp::FinalizeExactAccumulator { .. }, + ) => kept(), + _ => BTreeSet::new(), + } +} + /// The groups (or partitions) a `by (keys)` over rows of schema `input` /// from `sources` is priced at, from declared or derivable facts only: the /// smallest of -/// - the product of each key's distinct-value count, which each source -/// read declares in `distinct_values_per_label` (summed over the -/// sources: a union holds at most the sum, while a join matching on the -/// key is over-counted), when every key has one; +/// - when the keys read one source and are all in +/// `unchanged_input_columns` (one source row's labels, unchanged), the +/// smallest product of the counts it declares in +/// `distinct_values_per_label` and `distinct_values_per_label_set` that +/// covers the keys +/// ([`SourceFacts::declared_distinct_combinations`](asap_types::workload::SourceFacts::declared_distinct_combinations)); +/// otherwise (several sources, a join pairing labels of different rows, +/// or a relabeled key) the product of each key's +/// `distinct_values_per_label` count summed over the sources (a union +/// holds at most the sum, while a join matching on the key is +/// over-counted), when every key has one; /// - the series (PromQL) or groups of rows (SQL) the sources declare in /// `series_count`, when they all do (their product over several /// sources): each is in one group; @@ -2906,14 +3051,17 @@ fn group_count_bound( keys: &GroupKeys, input: Option<&Schema>, sources: &BTreeSet<&str>, + unchanged_input_columns: &BTreeSet<&str>, data: &DataWorkload, rows_read: u64, ) -> u64 { - let distinct_values = |key: &usize| -> Option { - let label = &input?.fields.get(*key)?.name; - if sources.is_empty() { - return None; - } + let labels = || -> Option> { + keys.keys() + .iter() + .map(|key| Some(input?.fields.get(*key)?.name.as_str())) + .collect() + }; + let label_count = |label: &str| -> Option { sources.iter().try_fold(0u64, |total, source| { let count = data .sources @@ -2923,11 +3071,26 @@ fn group_count_bound( Some(total.saturating_add(*count)) }) }; - let declared = if keys.is_without() { - None - } else { - keys.keys().iter().try_fold(1u64, |product, key| { - Some(product.saturating_mul(distinct_values(key)?)) + let declared = || -> Option { + if keys.is_without() || sources.is_empty() { + return None; + } + let labels = labels()?; + // Keys that are one source row's unchanged labels: the source's + // declared label sets bound their combinations. + if let (1, Some(source)) = (sources.len(), sources.first()) { + if labels + .iter() + .all(|label| unchanged_input_columns.contains(label)) + { + return data + .sources + .get(*source)? + .declared_distinct_combinations(&labels); + } + } + labels.into_iter().try_fold(1u64, |product, label| { + Some(product.saturating_mul(label_count(label)?)) }) }; // Over several sources a group combines one series (or group of rows) @@ -2939,7 +3102,7 @@ fn group_count_bound( Some(product.saturating_mul(data.sources.get(*source)?.series_count?)) }) }; - [declared, declared_series, Some(rows_read)] + [declared(), declared_series, Some(rows_read)] .into_iter() .flatten() .min() @@ -3238,6 +3401,9 @@ fn price_at_scale( }; // The sources each node reads, filled children first. let mut sources_read_by: HashMap> = HashMap::new(); + // Each node's output columns that still hold one source row's label (or + // column) values unchanged, by name ([`unchanged_source_columns`]). + let mut unchanged_columns_of: HashMap> = HashMap::new(); let multipliers = computations( dag, recurrence, @@ -3307,6 +3473,8 @@ fn price_at_scale( ) }; sources_read_by.insert(node.id, sources); + let unchanged = unchanged_source_columns(node, producers, nodes, &unchanged_columns_of); + unchanged_columns_of.insert(node.id, unchanged); let family = match &node.payload { Payload::ASAP( ASAPOp::SummaryAgg { @@ -3410,6 +3578,9 @@ fn price_at_scale( keys, input_schema, &sources_read_by[&node.id], + producers + .first() + .map_or(&BTreeSet::new(), |producer| &unchanged_columns_of[producer]), data, rows_read, )) @@ -4846,7 +5017,7 @@ mod tests { use asap_types::ir::scalar::expr_ir::ArithmeticOpKind; use asap_types::ir::schema::{GroupingStrategy, SketchKind}; use asap_types::ir::{ExprSemantics, ScalarExpr}; - use asap_types::workload::{DataArrival, MetricType, SourceFacts}; + use asap_types::workload::{DataArrival, LabelSetDistinctValues, MetricType, SourceFacts}; /// The AWS Fargate preset holds the page's prices converted as its doc /// says at the given speed: per vCPU-second × seconds per operation or @@ -6417,6 +6588,111 @@ mod tests { ); } + /// `x` holding 10,000 series over 4 regions, 25 services and 625 + /// endpoints, each endpoint in one service (the eval's RollupHeavy + /// `http`), with `(service, endpoint)`'s 625 combinations declared when + /// `declared_set`. + fn hierarchical_labels(declared_set: bool) -> DataWorkload { + let mut data = with_series(Some(10_000)); + let facts = data.sources.get_mut("x").unwrap(); + facts.distinct_values_per_label = [ + ("region".to_string(), 4), + ("service".to_string(), 25), + ("endpoint".to_string(), 625), + ] + .into(); + if declared_set { + facts.distinct_values_per_label_set = vec![LabelSetDistinctValues { + labels: vec!["service".to_string(), "endpoint".to_string()], + distinct_values: 625, + }]; + } + data + } + + /// A grouping is priced at the smallest bound the declared facts give: + /// the declared `(service, endpoint)` set's 625 combinations, not + /// 25 x 625 capped at the 10,000 series; a superset grouping at the set + /// times `region`'s 4; without the set, at the per-label product capped + /// at the series. A partitioned top-k takes the same count. + #[test] + fn a_grouping_is_priced_at_its_declared_label_set_count() { + let details = |query: &str, declared_set: bool| { + let data = hierarchical_labels(declared_set); + let cost = price( + &exact_dag(&[query], &data), + &recurrences(1), + &data, + &PlanSelectionSettings::default(), + ) + .unwrap(); + node_prices(&cost) + .into_iter() + .map(|(_, _, detail, _)| detail) + .collect::>() + }; + let sum = |keys: &str| format!("sum by ({keys}) (sum_over_time(x[15m]))"); + for (query, declared_set, expected) in [ + (sum("service, endpoint"), true, "into 625 states"), + (sum("region, service, endpoint"), true, "into 2500 states"), + ( + sum("service, endpoint"), + false, + "Sum accumulator into 10000 states", + ), + (sum("region, service"), true, "into 100 states"), + ( + "topk by (service, endpoint) (3, sum_over_time(x[15m]))".to_string(), + true, + " in 625 partitions;", + ), + // A relabeled key no longer holds the declared set's values: + // `endpoint` now joins `region` and `endpoint` (up to 4 x 625). + ( + r#"sum by (service, endpoint) (label_join(sum_over_time(x[15m]), "endpoint", "-", "region", "endpoint"))"#.to_string(), + true, + "into 10000 states", + ), + // A binary operator pairs the labels of two rows: per-label only. + ( + "sum by (service, endpoint) (sum_over_time(x[15m]) / on (service) group_left (endpoint) sum_over_time(x[15m]))".to_string(), + true, + "into 10000 states", + ), + ] { + let details = details(&query, declared_set); + assert!( + details.iter().any(|d| d.contains(expected)), + "{query} (set declared: {declared_set}): {details:#?}" + ); + } + } + + /// A grouped root's `G` takes the declared label set's count too: the + /// set's 625 for `(service, endpoint)`, times `region`'s 4 for a + /// superset; without the set, the per-label product capped at the + /// series. + #[test] + fn values_per_result_is_bound_by_a_declared_label_set() { + let root = + |query: &str| QueryRoot::Operator(lower_promql(query, AccuracyTarget::max_error(0.01))); + for (query, declared_set, values) in [ + ("sum by (service, endpoint) (rate(x[5m]))", true, 625), + ( + "sum by (region, service, endpoint) (rate(x[5m]))", + true, + 2_500, + ), + ("sum by (service, endpoint) (rate(x[5m]))", false, 10_000), + ] { + assert_eq!( + values_of(&root(query), &hierarchical_labels(declared_set)), + Some(values), + "{query} (set declared: {declared_set})" + ); + } + } + /// A grouped root's `G` is also bound by its keys' declared distinct /// values, at most the series read: `by (job)` with 10 jobs returns at /// most 10 values, also below a sort; a key without a declared count, a diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs index 00e9dc72b..7c86cdb39 100644 --- a/crates/types/src/workload/mod.rs +++ b/crates/types/src/workload/mod.rs @@ -577,6 +577,22 @@ pub struct SourceFacts { /// NULL) as one more value, and positive. Omitted from JSON when empty. #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub distinct_values_per_label: BTreeMap, + /// How many distinct combinations sets of labels (or columns) take + /// together, for labels that depend on each other (default: none + /// declared), e.g. `[{"labels": ["service", "endpoint"], + /// "distinct_values": 625}]` when each of 625 endpoints is in one of 25 + /// services. Plan selection prices a grouping at the smallest product + /// of declared counts whose labels cover its keys + /// ([`SourceFacts::declared_distinct_combinations`]), so + /// `by (service, endpoint)` is 625 groups rather than 25 x 625, and + /// `by (region, service, endpoint)` is at most 4 x 625. Nothing is + /// inferred from the per-label counts. Each set names two or more + /// distinct labels, and its count is a positive upper bound (a missing + /// label or NULL counts as one more value) no larger than the product + /// of its labels' `distinct_values_per_label` counts when all are + /// declared. Omitted from JSON when empty. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub distinct_values_per_label_set: Vec, /// How many rows (PromQL samples) the source holds at rest, at most /// (default: unknown). Plan selection bounds the rows a scan reads by /// it: the whole source for a scan with no window above it, such as a @@ -586,6 +602,81 @@ pub struct SourceFacts { pub rows_held: Option, } +/// The distinct combinations of values a set of labels (or columns) of one +/// source takes together, an upper bound. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LabelSetDistinctValues { + /// The labels (PromQL) or columns (SQL), two or more, in any order. + pub labels: Vec, + /// How many distinct combinations of their values the source holds. + pub distinct_values: u64, +} + +/// The most keys declared label sets are searched over: covering them +/// searches every subset of the keys some declared set names. +const MAX_KEYS_COVERED_BY_LABEL_SETS: usize = 16; + +impl SourceFacts { + /// An upper bound on the distinct combinations of values `labels` take + /// together in this source, from declared counts only: the smallest + /// product of counts, of declared label sets + /// (`distinct_values_per_label_set`) and of single labels + /// (`distinct_values_per_label`), whose labels together cover + /// `labels`. A set covering labels beyond `labels` still bounds them. + /// `Some(1)` for no labels; `None` when some label is in no declared + /// count. Only labels some declared set names are searched over (every + /// other label contributes its own count); with more than 16 of them, + /// only the per-label product is used. + pub fn declared_distinct_combinations(&self, labels: &[&str]) -> Option { + let mut keys = labels.to_vec(); + keys.sort_unstable(); + keys.dedup(); + let label_product = |keys: &[&str]| { + keys.iter().try_fold(1u64, |product, key| { + Some(product.saturating_mul(*self.distinct_values_per_label.get(*key)?)) + }) + }; + let (in_sets, alone): (Vec<&str>, Vec<&str>) = keys.iter().partition(|key| { + self.distinct_values_per_label_set + .iter() + .any(|set| set.labels.iter().any(|label| label == *key)) + }); + if in_sets.is_empty() || in_sets.len() > MAX_KEYS_COVERED_BY_LABEL_SETS { + return label_product(&keys); + } + // Each declared count over `in_sets` as (the keys it covers as a bit + // mask, count). + let mut counts: Vec<(usize, u64)> = in_sets + .iter() + .enumerate() + .filter_map(|(bit, key)| Some((1 << bit, *self.distinct_values_per_label.get(*key)?))) + .collect(); + counts.extend(self.distinct_values_per_label_set.iter().filter_map(|set| { + let covered = in_sets + .iter() + .enumerate() + .filter(|(_, key)| set.labels.iter().any(|label| label == *key)) + .fold(0, |mask, (bit, _)| mask | 1 << bit); + (covered != 0).then_some((covered, set.distinct_values)) + })); + // smallest[m]: the smallest product of counts covering the keys in m. + let every_key = (1usize << in_sets.len()) - 1; + let mut smallest: Vec> = vec![None; every_key + 1]; + smallest[0] = Some(1); + for keys_left in 1..=every_key { + smallest[keys_left] = counts + .iter() + .filter(|(covered, _)| covered & keys_left != 0) + .filter_map(|(covered, count)| { + Some(smallest[keys_left & !covered]?.saturating_mul(*count)) + }) + .min(); + } + Some(smallest[every_key]?.saturating_mul(label_product(&alone)?)) + } +} + /// Upper bounds on the norms of the frequency (weight) vector any one /// summary evaluation over a source folds. A filter or a shorter window only /// lowers them, so a bound for the whole source holds for every summary @@ -693,7 +784,7 @@ impl DataWorkload { /// or language. pub fn validate(&self) -> Result<(), WorkloadError> { validate_optional_rate(self.arrival.rows_per_second().map(Rate))?; - for facts in self.sources.values() { + for (source_name, facts) in &self.sources { if facts.series_count == Some(0) { return Err(WorkloadError::ZeroSeriesCount); } @@ -720,6 +811,32 @@ impl DataWorkload { if facts.rows_held == Some(0) { return Err(WorkloadError::ZeroRowsHeld); } + for set in &facts.distinct_values_per_label_set { + let error_labels = || set.labels.clone(); + let mut distinct_labels: Vec<&String> = set.labels.iter().collect(); + distinct_labels.sort_unstable(); + distinct_labels.dedup(); + if distinct_labels.len() < 2 || distinct_labels.len() != set.labels.len() { + return Err(WorkloadError::LabelSetNeedsTwoLabels { + source_name: source_name.clone(), + labels: error_labels(), + }); + } + if set.distinct_values == 0 { + return Err(WorkloadError::ZeroDistinctValues); + } + let label_product = set.labels.iter().try_fold(1u64, |product, label| { + Some(product.saturating_mul(*facts.distinct_values_per_label.get(label)?)) + }); + if let Some(label_product) = label_product.filter(|&p| set.distinct_values > p) { + return Err(WorkloadError::LabelSetCountAboveLabelProduct { + source_name: source_name.clone(), + labels: error_labels(), + distinct_values: set.distinct_values, + label_product, + }); + } + } } Ok(()) } @@ -802,7 +919,7 @@ fn validate_rate(rate: Rate) -> Result { } /// Why a planning workload is rejected. Every variant is bad input. -#[derive(Debug, Clone, Copy, PartialEq, thiserror::Error)] +#[derive(Debug, Clone, PartialEq, thiserror::Error)] pub enum WorkloadError { /// Bad input: a PromQL or MetricsQL query has a bare instant selector, /// which reads the latest sample within its metric's sample interval, @@ -865,10 +982,37 @@ pub enum WorkloadError { /// Bad input: a source's `max_distinct_per_evaluation` of zero. #[error("a source's max_distinct_per_evaluation must be greater than zero")] ZeroMaxDistinct, - /// Bad input: a label (or column) declared with zero distinct values in - /// a source's `distinct_values_per_label`. - #[error("a source's distinct_values_per_label counts must be greater than zero")] + /// Bad input: a zero count in a source's `distinct_values_per_label` + /// or `distinct_values_per_label_set`. + #[error( + "a source's distinct_values_per_label and distinct_values_per_label_set counts must be \ + greater than zero" + )] ZeroDistinctValues, + /// Bad input: a `distinct_values_per_label_set` entry naming fewer than + /// two labels, or one label twice (one label's count belongs in + /// `distinct_values_per_label`). + #[error( + "source {source_name}: a distinct_values_per_label_set entry must name two or more \ + distinct labels, got {labels:?}" + )] + LabelSetNeedsTwoLabels { + source_name: String, + labels: Vec, + }, + /// Bad input: a label set declaring more combinations than the product + /// of its labels' declared distinct values allows. + #[error( + "source {source_name}: distinct_values_per_label_set declares {distinct_values} \ + combinations of {labels:?}, more than the {label_product} their \ + distinct_values_per_label counts allow" + )] + LabelSetCountAboveLabelProduct { + source_name: String, + labels: Vec, + distinct_values: u64, + label_product: u64, + }, /// Bad input: a source's `rows_held` of zero. #[error("a source's rows_held must be greater than zero")] ZeroRowsHeld, @@ -1031,6 +1175,128 @@ mod tests { assert_eq!(with(0).validate(), Err(WorkloadError::ZeroDistinctValues)); } + /// The `http` source of the eval's RollupHeavy workload: 4 regions, 25 + /// services and 625 endpoints, each endpoint in one service, so + /// (service, endpoint) takes 625 combinations, not 25 x 625. + fn hierarchical_facts(declared_sets: &[(&[&str], u64)]) -> SourceFacts { + SourceFacts { + distinct_values_per_label: BTreeMap::from([ + ("region".to_string(), 4), + ("service".to_string(), 25), + ("endpoint".to_string(), 625), + ]), + distinct_values_per_label_set: declared_sets + .iter() + .map(|(labels, distinct_values)| LabelSetDistinctValues { + labels: labels.iter().map(|label| label.to_string()).collect(), + distinct_values: *distinct_values, + }) + .collect(), + ..SourceFacts::default() + } + } + + /// A grouping's combinations are bounded by the smallest product over + /// declared sets and labels covering its keys: the declared set itself, + /// a superset of keys through the set times the other labels, and a + /// subset of a declared set through that set. Without a declared set, + /// the per-label product stands. + #[test] + fn declared_label_sets_bound_the_combinations_of_their_labels() { + let declared = hierarchical_facts(&[(&["service", "endpoint"], 625)]); + let undeclared = hierarchical_facts(&[]); + let combinations = + |facts: &SourceFacts, labels: &[&str]| facts.declared_distinct_combinations(labels); + assert_eq!(combinations(&declared, &["service", "endpoint"]), Some(625)); + assert_eq!(combinations(&declared, &["endpoint", "service"]), Some(625)); + assert_eq!( + combinations(&declared, &["region", "service", "endpoint"]), + Some(4 * 625) + ); + assert_eq!( + combinations(&undeclared, &["service", "endpoint"]), + Some(25 * 625) + ); + assert_eq!( + combinations(&undeclared, &["region", "service", "endpoint"]), + Some(4 * 25 * 625) + ); + assert_eq!(combinations(&declared, &["service"]), Some(25)); + assert_eq!(combinations(&declared, &[]), Some(1)); + // A key no fact covers leaves the combinations unbound. + assert_eq!(combinations(&declared, &["service", "status"]), None); + // A declared set covering more labels bounds its subsets: 3 users + // have made 7 (user, session) pairs, so at most 7 sessions, while + // `session` alone declares no count. + let sessions = SourceFacts { + distinct_values_per_label_set: vec![LabelSetDistinctValues { + labels: vec!["user".to_string(), "session".to_string()], + distinct_values: 7, + }], + ..SourceFacts::default() + }; + assert_eq!( + sessions.declared_distinct_combinations(&["session"]), + Some(7) + ); + // 70 keys, two in a declared set: only those two are searched over. + let mut wide = hierarchical_facts(&[(&["service", "endpoint"], 625)]); + let names: Vec = (0..68).map(|i| format!("l{i}")).collect(); + for name in &names { + wide.distinct_values_per_label.insert(name.clone(), 1); + } + let mut labels: Vec<&str> = names.iter().map(String::as_str).collect(); + labels.extend(["service", "endpoint"]); + assert_eq!(wide.declared_distinct_combinations(&labels), Some(625)); + } + + /// A declared set count is an input error, naming the source, when it + /// is above the product of its labels' declared counts, or when the set + /// has fewer than two distinct labels (one label's count belongs in + /// `distinct_values_per_label`) or zero values. + #[test] + fn a_label_set_count_is_at_most_its_labels_product() { + let with = |facts: SourceFacts| DataWorkload { + sources: BTreeMap::from([("http".to_string(), facts)]), + ..DataWorkload::default() + }; + assert!(with(hierarchical_facts(&[(&["service", "endpoint"], 625)])) + .validate() + .is_ok()); + // `status` declares no count, so the set's product is unknown. + assert!(with(hierarchical_facts(&[(&["service", "status"], 1_000)])) + .validate() + .is_ok()); + let error = with(hierarchical_facts(&[(&["region", "service"], 101)])) + .validate() + .unwrap_err(); + assert_eq!( + error, + WorkloadError::LabelSetCountAboveLabelProduct { + source_name: "http".to_string(), + labels: vec!["region".to_string(), "service".to_string()], + distinct_values: 101, + label_product: 100, + } + ); + assert!(error.to_string().contains("http"), "{error}"); + for labels in [&["service"][..], &["service", "service"][..]] { + let error = with(hierarchical_facts(&[(labels, 25)])).validate(); + assert!( + matches!( + &error, + Err(WorkloadError::LabelSetNeedsTwoLabels { source_name, .. }) + if source_name == "http" + ), + "{error:?}" + ); + } + assert_eq!( + with(hierarchical_facts(&[(&["service", "endpoint"], 0)])).validate(), + Err(WorkloadError::ZeroDistinctValues) + ); + } + /// A declared series count is positive: with zero series, `absent(m)` /// would still return one row the bound does not count. #[test] @@ -1300,7 +1566,14 @@ mod tests { l2: None, }), max_distinct_per_evaluation: Some(100), - distinct_values_per_label: BTreeMap::from([("job".to_string(), 10)]), + distinct_values_per_label: BTreeMap::from([ + ("job".to_string(), 10), + ("instance".to_string(), 1_000), + ]), + distinct_values_per_label_set: vec![LabelSetDistinctValues { + labels: vec!["job".to_string(), "instance".to_string()], + distinct_values: 1_000, + }], rows_held: Some(1_000_000), }, )]), @@ -1319,7 +1592,10 @@ mod tests { "sample_interval": 15000, "stream_norms": {"l1": 1e9, "l2": null}, "max_distinct_per_evaluation": 100, - "distinct_values_per_label": {"job": 10}, + "distinct_values_per_label": {"instance": 1000, "job": 10}, + "distinct_values_per_label_set": [ + {"labels": ["job", "instance"], "distinct_values": 1000} + ], "rows_held": 1000000 }} }) diff --git a/docs/design_docs/architecture/input-output-workflow.md b/docs/design_docs/architecture/input-output-workflow.md index 6af8ad8cf..6de4c9a82 100644 --- a/docs/design_docs/architecture/input-output-workflow.md +++ b/docs/design_docs/architecture/input-output-workflow.md @@ -201,8 +201,13 @@ struct SourceFacts { stream_norms: Option, max_distinct_per_evaluation: Option, distinct_values_per_label: BTreeMap, + distinct_values_per_label_set: Vec, rows_held: Option, } +struct LabelSetDistinctValues { + labels: Vec, + distinct_values: u64, +} ``` | Field | Required | Purpose and behavior when unavailable | @@ -215,6 +220,7 @@ struct SourceFacts { | `sample_interval` | PromQL and MetricsQL; nonzero | How often each series of the metric is sampled. Bare instant selectors over it use it as their selection horizon; without it, such a query fails to lower. | | `stream_norms`, `max_distinct_per_evaluation` | Optional | Bounds that Count-Min/Count Sketch and HLL/UnivMon guarantees are bound with. | | `distinct_values_per_label` | Optional; default empty; each count nonzero | Distinct values each label (PromQL) or column (SQL) takes, e.g. `{"host": 1000, "job": 10}`. Plan selection prices a `by (k1, k2, ...)` aggregate or summary, and a partitioned sort or limit, at the product of its keys' counts, at most the `series_count` (series, or groups of rows) of the sources it reads and the rows it reads per evaluation. Over several sources a key's counts add. No count is assumed for a key some source read does not declare: the grouping is priced at the smaller of that `series_count` and those rows. Each count must be an upper bound, counting a missing label (or a NULL) as one more value: when every key of a grouped result declares one, their product (at most the series read) also bounds `G`, which sizes per-group sketches for a whole-result target; otherwise `G` keeps its series bound, or none for SQL. | +| `distinct_values_per_label_set` | Optional; default empty; each set two or more distinct labels, its count nonzero and at most the product of its labels' declared counts | Distinct combinations a set of labels (or columns) takes together, for labels that depend on each other, e.g. `[{"labels": ["service", "endpoint"], "distinct_values": 625}]` when each of 625 endpoints is in one of 25 services. For a grouping over one source whose keys are one row's unchanged labels, the per-label product above becomes the smallest product of declared set and label counts whose labels cover the keys: `by (service, endpoint)` is 625 groups, not 25 x 625, and `by (region, service, endpoint)` at most 4 x 625. A set covering more labels than the keys also bounds them. Nothing is inferred from the per-label counts (no functional dependency is assumed). Over several sources, or for keys a join or binary operator pairs from different rows, only the per-label counts apply; a key a relabeling rewrites is not bounded by any declared set. The same bound gives `G` for a whole-result target. A count above its labels' product is an input error naming the source. | | `rows_held` | Optional; nonzero | Rows the source holds at rest. A query-time scan reads at most these; a scan with no window above it (a SQL query without a time predicate) reads all of them. | Every fact is declared: the planner has no clock, so a caller drops facts diff --git a/docs/design_docs/concepts/accuracy-models.md b/docs/design_docs/concepts/accuracy-models.md index 9e0868129..c28ec395f 100644 --- a/docs/design_docs/concepts/accuracy-models.md +++ b/docs/design_docs/concepts/accuracy-models.md @@ -84,8 +84,12 @@ strictest: every group (`G`, when `G` is above 100), then a 1% share (100), then a 5% share (20), then a single value (1). A whole-result target with an unbound `G` enlarges nothing and stays unmet. A grouped result's `G` is bound by the distinct values its sources declare for its group keys -(`SourceFacts.distinct_values_per_label`): the product over the keys, at most -the series read. A key counts only when it is a source column passed through +(`SourceFacts.distinct_values_per_label`, and +`distinct_values_per_label_set` for labels that depend on each other): the +smallest product of declared label and label-set counts covering the keys, at +most the series read. With each of 625 endpoints in one of 25 services and +`(service, endpoint)` declared at 625, `by (service, endpoint)` returns at +most 625 values, not 25 x 625. A key counts only when it is a source column passed through unchanged; a relabeled or computed key leaves `G` unbound. So a sketch meets a whole-result target over SQL groups when every grouping column's distinct values are declared, and over 25 services and 25 endpoints sizes each group's sketch for `δ / 625`. Nothing declared diff --git a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md index 33739b0f4..c5a5a6087 100644 --- a/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md +++ b/docs/design_docs/proposals/asap-aware-mapping/analytical-resource-cost.md @@ -231,7 +231,7 @@ normalized workload, lowered query IR, and declared statistics: | Cost concept | Canonical source | |---|---| | Source row cardinality | `DataWorkload.sources[source].series_count`, and `rows_held` for data at rest or a scan with no window. No default: a missing fact fails planning, naming the source and the fact. | -| Groups of a `by (...)` grouping or partition | The smallest fact-based bound: the product of `DataWorkload.sources[source].distinct_values_per_label` over its keys (all declared), the declared `series_count` (series, or groups of rows for a SQL table) of the sources it reads, and the rows it reads per evaluation (at ingestion time, the arriving rows of the window it maintains; a grouping with no window is an error). No count is assumed for an undeclared key. | +| Groups of a `by (...)` grouping or partition | The smallest fact-based bound: the product of `DataWorkload.sources[source].distinct_values_per_label` over its keys (all declared), or over one source whose keys are one row's unchanged labels (no join, binary operator or relabeling below), the smallest product of its declared label and `distinct_values_per_label_set` counts covering the keys, the declared `series_count` (series, or groups of rows for a SQL table) of the sources it reads, and the rows it reads per evaluation (at ingestion time, the arriving rows of the window it maintains; a grouping with no window is an error). No count is assumed for an undeclared key. | | Arrival and update rate | `DataWorkload.arrival` and its `rows_per_second`. | | Query demand | `QueryWorkloadEntry.recurrence`, normalized over an explicit horizon. | | Event-time range | The query text (its range and offset). | diff --git a/docs/develop_docs/library-api.md b/docs/develop_docs/library-api.md index 7de4ade55..d446db5d0 100644 --- a/docs/develop_docs/library-api.md +++ b/docs/develop_docs/library-api.md @@ -298,7 +298,12 @@ plan selection prices a `by (...)` grouping's groups at: the product of its keys' counts, at most the declared `series_count` (series or groups of rows) and the rows it reads; with a key undeclared, no count is assumed and the grouping is priced at the smaller of those two; over several sources a -key's counts add), and `rows_held` (the rows the source holds at rest, +key's counts add), `distinct_values_per_label_set` (distinct combinations +a set of dependent labels takes together, e.g. +`[{"labels": ["service", "endpoint"], "distinct_values": 625}]` when each +endpoint is in one service: a grouping over one source is priced at the +smallest product of declared set and label counts covering its keys, so +`by (service, endpoint)` is 625 groups rather than 25 x 625), and `rows_held` (the rows the source holds at rest, which bound a scan with no window and data at rest). Every fact is declared: the planner has no clock, so a caller drops facts that no longer hold before planning, and pricing assumes no default: a scan whose source declares no arrival rate diff --git a/docs/user_guide_docs/stage-viewer.md b/docs/user_guide_docs/stage-viewer.md index b9a8119b8..0ff7e4278 100644 --- a/docs/user_guide_docs/stage-viewer.md +++ b/docs/user_guide_docs/stage-viewer.md @@ -127,12 +127,15 @@ planner must meet: Beside the queries, the facts the planner was given about the data: how it arrives (`ingesting · 200000 rows/s`, `at rest`, `mixed`) and, per source (a PromQL metric or a SQL table), its series count, metric type, whether -its values are non-negative, its sample interval, its stream-norm bounds and -its distinct-count bound, `—` for a fact the workload does not declare. The +its values are non-negative, its sample interval, its stream-norm bounds, +its distinct-count bound, the distinct values per label (`service 25, +endpoint 625`) and per label set (`(service, endpoint) 625`, for labels that +depend on each other, such as endpoints that each belong to one service), +`—` for a fact the workload does not declare. The planner prices a grouping +at the smallest product of these counts covering its keys. The **read by** column lists the queries whose text names the source. The line above the table shows the deployment capabilities the plan was made for -(for example `raw data not retained`). A workload cannot declare labels, -their hierarchy or their cardinalities, so the panel shows none. +(for example `raw data not retained`). The line under the queries counts the candidates at each stage, for example, in Example 1, `Candidates per stage: 1 logical DAG → 27 logical ASAP → 72 diff --git a/tools/dag-viewer/README.md b/tools/dag-viewer/README.md index 0a88ca51e..92ca6f020 100644 --- a/tools/dag-viewer/README.md +++ b/tools/dag-viewer/README.md @@ -40,7 +40,7 @@ To run it, see [RUNNING.md](RUNNING.md). - **Data workload** beside it: the arrival and one row per declared source with exactly the facts `SourceFacts` has (series count, metric type, non-negative values, sample interval, stream norms, distinct bound, - distinct values per label) and + distinct values per label, distinct values per label set) and the queries whose text names it, plus the deployment capabilities the run planned with. - **Stage 3 · plans by cost**: every candidate the document shows. First the diff --git a/tools/dag-viewer/app.js b/tools/dag-viewer/app.js index cb40f5cee..408647431 100644 --- a/tools/dag-viewer/app.js +++ b/tools/dag-viewer/app.js @@ -297,8 +297,8 @@ $('dataArrival').textContent = `arrival: ${view.arrival}`; $('deployment').textContent = doc.deployment_capabilities ? `Deployment: ${deploymentText(doc.deployment_capabilities)}` : ''; $('dataSources').innerHTML = view.rows.length - ? table(['source', 'series', 'type', 'non-negative', 'sample interval', 'stream norms', 'distinct bound', 'distinct values per label', 'read by'], - view.rows.map((r) => `${esc(r.name)}${esc(r.series)}${esc(r.type)}${esc(r.nonNegative)}${esc(r.sampleInterval)}${esc(r.streamNorms)}${esc(r.maxDistinct)}${esc(r.distinctValuesPerLabel)}${esc(r.readBy.join(', ') || '—')}`).join('')) + ? table(['source', 'series', 'type', 'non-negative', 'sample interval', 'stream norms', 'distinct bound', 'distinct values per label', 'distinct values per label set', 'read by'], + view.rows.map((r) => `${esc(r.name)}${esc(r.series)}${esc(r.type)}${esc(r.nonNegative)}${esc(r.sampleInterval)}${esc(r.streamNorms)}${esc(r.maxDistinct)}${esc(r.distinctValuesPerLabel)}${esc(r.distinctValuesPerLabelSet)}${esc(r.readBy.join(', ') || '—')}`).join('')) : 'No source declared.'; } diff --git a/tools/dag-viewer/stages.js b/tools/dag-viewer/stages.js index 681f3b255..4913c2b16 100644 --- a/tools/dag-viewer/stages.js +++ b/tools/dag-viewer/stages.js @@ -809,6 +809,7 @@ function dataWorkloadView(dataWorkload, queries) { streamNorms: norms(facts.stream_norms), maxDistinct: typeof facts.max_distinct_per_evaluation === 'number' ? facts.max_distinct_per_evaluation : '—', distinctValuesPerLabel: Object.entries(facts.distinct_values_per_label || {}).map(([label, count]) => `${label} ${count}`).join(', ') || '—', + distinctValuesPerLabelSet: (Array.isArray(facts.distinct_values_per_label_set) ? facts.distinct_values_per_label_set : []).map((set) => `(${(set.labels || []).join(', ')}) ${set.distinct_values}`).join(', ') || '—', readBy: queriesNaming(name, queries || []), })); return { arrival: `${kind}${rate}`, rows }; diff --git a/tools/dag-viewer/test_viewer.py b/tools/dag-viewer/test_viewer.py index 973d5f397..b175076c8 100644 --- a/tools/dag-viewer/test_viewer.py +++ b/tools/dag-viewer/test_viewer.py @@ -539,6 +539,15 @@ def test_data_workload_rows_show_each_labels_distinct_values(self): self.assertEqual(rows["data"]["distinctValuesPerLabel"], "host 1000, job 10") self.assertEqual(rows["other"]["distinctValuesPerLabel"], "—") + def test_data_workload_rows_show_each_label_sets_distinct_values(self): + data_workload = {"arrival": "at_rest", "sources": { + "http": {"series_count": 10000, "distinct_values_per_label": {"service": 25, "endpoint": 625}, + "distinct_values_per_label_set": [{"labels": ["service", "endpoint"], "distinct_values": 625}]}, + "other": {"series_count": 5}}} + rows = {r["name"]: r for r in self.js.call("dataWorkloadView", data_workload, [])["rows"]} + self.assertEqual(rows["http"]["distinctValuesPerLabelSet"], "(service, endpoint) 625") + self.assertEqual(rows["other"]["distinctValuesPerLabelSet"], "—") + def test_stored_output_rows_map_to_stage2_nodes_and_their_readers(self): rows = self.js.call("storedOutputRows", self.doc) stored = self.doc["stage3_plan_selection"]["stored_outputs"]