From e568ac86d7ece1e3b1173e342218ba1d5a17c8d5 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 6 Nov 2026 13:40:37 +0000 Subject: [PATCH 1/3] refactor(types): properties and requirements from #1207 (#1207 PR 1) --- crates/accuracy/src/estimators/cms.rs | 1 + .../accuracy/src/estimators/count_sketch.rs | 1 + crates/accuracy/src/estimators/ddsketch.rs | 1 + crates/accuracy/src/estimators/hll.rs | 1 + crates/accuracy/src/estimators/mod.rs | 181 +++- crates/accuracy/src/estimators/univmon.rs | 7 + crates/accuracy/src/sizing.rs | 28 +- crates/accuracy/src/test_support.rs | 20 +- crates/types/src/physical/properties.rs | 61 +- crates/types/src/stages/candidates.rs | 837 +++++++++++++++++- .../src/stages/physical/memo_candidates.rs | 325 ++++++- crates/types/src/workload/mod.rs | 69 +- 12 files changed, 1425 insertions(+), 107 deletions(-) diff --git a/crates/accuracy/src/estimators/cms.rs b/crates/accuracy/src/estimators/cms.rs index ebce49b97..2efe584a7 100644 --- a/crates/accuracy/src/estimators/cms.rs +++ b/crates/accuracy/src/estimators/cms.rs @@ -25,6 +25,7 @@ pub(super) fn guarantee( } /// CMS: over-count ≤ ε·N with width `w = ⌈e/ε⌉` columns. +#[cfg(test)] pub(crate) fn cms_width(eps: f64) -> u32 { saturating_ceil(std::f64::consts::E / eps, 2, 1 << 26) } diff --git a/crates/accuracy/src/estimators/count_sketch.rs b/crates/accuracy/src/estimators/count_sketch.rs index 433803db3..56db95133 100644 --- a/crates/accuracy/src/estimators/count_sketch.rs +++ b/crates/accuracy/src/estimators/count_sketch.rs @@ -57,6 +57,7 @@ fn count_sketch_failure_probability(width: u32, depth: u32) -> Option { } /// CountSketch `L2` point-query width: ε = sqrt(3/w). +#[cfg(test)] pub(crate) fn count_sketch_width(eps: f64) -> u32 { saturating_ceil(3.0 / (eps * eps), 2, 1 << 26) } diff --git a/crates/accuracy/src/estimators/ddsketch.rs b/crates/accuracy/src/estimators/ddsketch.rs index 9cde75477..5f1a7310c 100644 --- a/crates/accuracy/src/estimators/ddsketch.rs +++ b/crates/accuracy/src/estimators/ddsketch.rs @@ -20,6 +20,7 @@ pub(super) fn guarantee( )) } +#[cfg(test)] pub(super) fn size_params(epsilon: f64) -> SketchParams { SketchParams::DDSketch { alpha: epsilon } } diff --git a/crates/accuracy/src/estimators/hll.rs b/crates/accuracy/src/estimators/hll.rs index 65ff8512f..09349cb4e 100644 --- a/crates/accuracy/src/estimators/hll.rs +++ b/crates/accuracy/src/estimators/hll.rs @@ -255,6 +255,7 @@ pub fn min_precision( /// bound up to `⌈1/ε⌉` (18 when none up to 18 does). Pass 1 then raises /// the precision by the same rule ([`min_precision`]) when the source's own /// bound or a smaller failure probability needs it. +#[cfg(test)] pub(crate) fn hll_precision(eps: f64, delta: f64) -> u8 { if !(eps.is_finite() && eps > 0.0 && delta.is_finite() && delta > 0.0) { return 18; diff --git a/crates/accuracy/src/estimators/mod.rs b/crates/accuracy/src/estimators/mod.rs index 69a14f3a1..75684c1e3 100644 --- a/crates/accuracy/src/estimators/mod.rs +++ b/crates/accuracy/src/estimators/mod.rs @@ -1,5 +1,6 @@ //! Dispatch committed estimator parameters to their accuracy models. use super::*; +#[cfg(test)] use asap_types::ir::operator::AggIntent; use asap_types::ir::schema::FamilyCapability; @@ -142,7 +143,10 @@ pub fn with_failure_probability(params: &SketchParams, delta: f64) -> SketchPara } } -#[allow(dead_code)] // Read again for the predefined θ lists (`parameter_list`), in #1207 PR 1. +/// Today's sizing: the parameters of family `kind` for error `eps` at failure +/// probability `delta`. Test-only: the predefined lists (`parameter_list`) +/// are this sizing over an error grid, which a test keeps equal. +#[cfg(test)] pub(crate) fn size_params( kind: SketchAlgorithm, intent: &AggIntent, @@ -249,6 +253,7 @@ pub(crate) fn binomial_upper_tail(n: u32, p: f64, k: u32) -> f64 { /// Accuracy-dependent candidate budget, not a completeness theorem. The /// membership model must still certify the selected set independently. +#[cfg(test)] pub(crate) fn topk_capacity(k: usize, eps: f64) -> u32 { u32::try_from(k) .unwrap_or(u32::MAX) @@ -269,6 +274,180 @@ mod tests { use super::*; use asap_types::ir::operator::agg_intent::{default_cardinality, default_quantile}; + /// Each family's predefined list is today's sizing at failure + /// probability 0.01 over the error grid, weakest first, without repeats + /// (a UnivMon's: its L2-norm, distinct-count and other shapes). KMV's and + /// UnivMon's distinct counts stop at ε = 0.01. A heap variant's width × depth is the top-1 + /// sizing, each with every heap size of the 1-2-5 series 10 … 10 000. + #[test] + fn parameter_list_is_the_sizing_over_the_error_grid() { + use asap_types::ir::properties::AccuracyTarget; + use asap_types::stages::parameter_list; + let full_grid = [0.1, 0.05, 0.02, 0.01, 0.005, 0.002, 0.001]; + let distinct_grid = [0.1, 0.05, 0.02, 0.01]; + let heap_sizes = [10, 20, 50, 100, 200, 500, 1_000, 2_000, 5_000, 10_000]; + let l2 = AggIntent::FrequencyL2 { + col: None, + accuracy: AccuracyTarget::max_error(0.01), + }; + let top_1 = AggIntent::TopK { + k: 1, + accuracy: AccuracyTarget::max_error(0.01), + }; + let quantile = default_quantile(0.5); + let distinct = default_cardinality(); + for (family, intents) in [ + (SketchAlgorithm::Kll, vec![&quantile]), + (SketchAlgorithm::DDSketch, vec![&quantile]), + (SketchAlgorithm::Hll, vec![&distinct]), + (SketchAlgorithm::Kmv, vec![&distinct]), + (SketchAlgorithm::Cms, vec![&distinct]), + (SketchAlgorithm::CountSketch, vec![&distinct]), + (SketchAlgorithm::CmsWithHeap, vec![&top_1]), + (SketchAlgorithm::CountSketchWithHeap, vec![&top_1]), + (SketchAlgorithm::UnivMon, vec![&l2, &distinct, &quantile]), + ] { + let mut sized: Vec = Vec::new(); + for intent in intents { + let grid: &[f64] = match (family, intent) { + ( + SketchAlgorithm::Kmv | SketchAlgorithm::UnivMon, + AggIntent::Cardinality { .. }, + ) => &distinct_grid, + _ => &full_grid, + }; + for &eps in grid { + let params = size_params(family, intent, eps, sizing::DEFAULT_DELTA); + let with_heaps: Vec = match params { + SketchParams::CmsWithHeap { width, depth, .. } => heap_sizes + .iter() + .map(|&heap_size| SketchParams::CmsWithHeap { + width, + depth, + heap_size, + }) + .collect(), + SketchParams::CountSketchWithHeap { width, depth, .. } => heap_sizes + .iter() + .map(|&heap_size| SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }) + .collect(), + other => vec![other], + }; + for params in with_heaps { + if !sized.contains(¶ms) { + sized.push(params); + } + } + } + } + assert_eq!( + format!("{:?}", parameter_list(family)), + format!("{sized:?}"), + "{family:?}" + ); + } + } + + /// Within each family's list, a configuration that satisfies another's + /// floor (`producer_satisfies_logical_requirements`) has no larger error, at the lists' failure + /// probability 0.01, for every statistic the family answers. + #[test] + fn a_configuration_satisfying_a_listed_floor_is_no_less_accurate() { + use asap_types::ir::operator::operator_properties::Reduction; + use asap_types::ir::schema::{GroupingStrategy, SummaryUpdate}; + use asap_types::ir::ASAPOp; + use asap_types::ir::Operator; + use asap_types::stages::{ + parameter_list, producer_satisfies_logical_requirements, CandidateExpression, + LogicalProperties, LogicalRequirements, MemoGroupId, + }; + let point = SketchStatistic::PointCount { + key: asap_types::ir::scalar::expr_ir::ColumnRef::SampleValue, + value: None, + }; + let statistics = [ + SketchStatistic::Quantile { q: 0.5 }, + SketchStatistic::Cardinality, + SketchStatistic::FrequencyL2, + SketchStatistic::FrequencyEntropy, + SketchStatistic::TopK { k: 1 }, + point, + ]; + for family in [ + SketchAlgorithm::Kll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::Hll, + SketchAlgorithm::Kmv, + SketchAlgorithm::Cms, + SketchAlgorithm::CountSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + SketchAlgorithm::UnivMon, + ] { + let build = CandidateExpression { + operator: Operator::ASAP(ASAPOp::SummaryAgg { + child: MemoGroupId(0), + family: FieldDataType::Sketch(family, GroupingStrategy::default()), + input: SummaryUpdate::column( + asap_types::ir::scalar::expr_ir::ColumnRef::SampleValue, + ), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }), + }; + let capability = + FamilyCapability::of(&FieldDataType::Sketch(family, GroupingStrategy::default())); + let list = parameter_list(family); + for producer in list { + let properties = + LogicalProperties::new(&build, Some(producer.clone()), None).unwrap(); + for floor in list { + let requirement = LogicalRequirements::new(Some(floor.clone())); + if !producer_satisfies_logical_requirements(&properties, &requirement) { + continue; + } + for statistic in statistics.iter().filter(|s| capability.answers(s)) { + let Some(needed) = sketch_guarantee(&family, floor, statistic) else { + continue; + }; + let delivered = sketch_guarantee(&family, producer, statistic) + .unwrap_or_else(|| panic!("{producer:?} {statistic:?}")); + // A bound that does not evaluate (an unknown term) is + // compared by its expression. + let error_no_larger = match ( + delivered.spec.error.evaluate(), + needed.spec.error.evaluate(), + ) { + (Some(error), Some(needed)) => error <= needed * (1.0 + 1e-9), + (_, _) => delivered.spec.error == needed.spec.error, + }; + let failure_no_likelier = match ( + delivered.spec.failure_probability.evaluate(), + needed.spec.failure_probability.evaluate(), + ) { + (Some(delta), Some(needed)) => { + delta <= needed.max(sizing::DEFAULT_DELTA) * (1.0 + 1e-9) + } + (_, _) => { + delivered.spec.failure_probability + == needed.spec.failure_probability + } + }; + assert!( + error_no_larger && failure_no_likelier, + "{producer:?} ⪰ {floor:?} for {statistic:?}: {delivered:?} vs {needed:?}" + ); + } + } + } + } + } + fn delta_and_error(params: &SketchParams, query: &SketchStatistic) -> (f64, f64) { let algorithm = match params { SketchParams::Cms { .. } => SketchAlgorithm::Cms, diff --git a/crates/accuracy/src/estimators/univmon.rs b/crates/accuracy/src/estimators/univmon.rs index 41b694d1a..85d6c9d5b 100644 --- a/crates/accuracy/src/estimators/univmon.rs +++ b/crates/accuracy/src/estimators/univmon.rs @@ -17,18 +17,22 @@ //! (OctoSketch aggregators fed promoted deltas) are not covered: the //! executor rejects them. use super::*; +#[cfg(test)] use asap_types::ir::operator::AggIntent; /// Layers `asap_sketchlib` builds: layer `i` hashes at seed index `i`, and /// seed 19 picks the layers a key reaches. +#[cfg(test)] pub(crate) const MAX_LAYERS: u8 = asap_sketchlib::sketch_framework::univmon::UNIVMON_MAX_LAYERS as u8; /// The most rows an L2 sketch is sized with: the deepest odd shape within /// the 20 rows the executor builds ([`hash_independent`]), so the row median /// is one row's estimate, and within the hash budget at the more than 20 /// columns an L2 guarantee needs (5 column bits: 19 · 6 = 114 ≤ 128). +#[cfg(test)] const MAX_L2_ROWS: u32 = 19; /// Each row of a layer reads its column and sign bits from one hash. +#[cfg(test)] const HASH_BITS: u32 = 128; /// The per-row failure probability the L2 guarantee fixes: a row misses by /// more than `ε₂ F2` with probability at most `2q/ε₂² = 0.1`. Fixing it @@ -50,6 +54,7 @@ const MAX_DISTINCT: &str = "max_distinct"; /// The shape baseline for readouts without a size model: the total (exact /// whatever the shape) and entropy (uncertified). +#[cfg(test)] const BASELINE: SketchParams = SketchParams::UnivMon { heap_size: 256, sketch_rows: 5, @@ -148,6 +153,7 @@ fn l2_failure_probability(rows: u32) -> f64 { /// clamped to the hash budget: with more rows than [`MAX_L2_ROWS`], or more /// columns than `rows` leave bits for, the shape is the largest one built /// and its guarantee misses the target. +#[cfg(test)] fn l2_shape(eps: f64, delta: f64) -> (u32, u32) { let rows = (1..=MAX_L2_ROWS) .find(|&rows| l2_failure_probability(rows) <= delta) @@ -347,6 +353,7 @@ pub(crate) fn with_failure_probability(params: &SketchParams, delta: f64) -> Ske /// counter; /// - total and entropy: the baseline shape (the total is exact; entropy /// has no certified bound). +#[cfg(test)] pub(super) fn size_params(intent: &AggIntent, eps: f64, delta: f64) -> SketchParams { match intent { AggIntent::FrequencyL2 { .. } => { diff --git a/crates/accuracy/src/sizing.rs b/crates/accuracy/src/sizing.rs index cdda5845c..23ef50cdd 100644 --- a/crates/accuracy/src/sizing.rs +++ b/crates/accuracy/src/sizing.rs @@ -7,7 +7,7 @@ use asap_types::ir::operator::AggIntent; use asap_types::ir::properties::{AccuracyMetric, AccuracyTarget}; use asap_types::ir::scalar::ColumnRef; -use asap_types::ir::schema::{SketchAlgorithm, SketchParams, SketchStatistic}; +use asap_types::ir::schema::SketchStatistic; /// Confidence δ assumed when the target carries only an ε /// (`AccuracyTarget::epsilon`): the (ε, δ)-parameterised sketches (CMS) need @@ -67,7 +67,7 @@ fn target_in(target: &AccuracyTarget, metric: AccuracyMetric) -> AccuracyTarget } /// Resolve an [`AccuracyTarget`] into the `(eps, delta)` budget -/// `default_size_params` needs — one place this resolution happens, so +/// today's sizing needs — one place this resolution happens, so /// nothing can drift apart on it. /// /// An exact target degrades to the tightest parameters for a caller that @@ -82,26 +82,10 @@ pub fn accuracy_budget(accuracy: &AccuracyTarget) -> (f64, f64) { ) } -/// `asap-plan`'s built-in `SketchParams` sizing, keyed off the resolved -/// `(eps, delta)` accuracy budget. -/// -/// Each formula inverts the sketch family's standard error bound to the -/// smallest parameter satisfying the target, clamped to the family's sane -/// range. A non-positive ε saturates to the clamp maximum (tightest -/// allowed). -#[allow(dead_code)] // Read by `test_support` until #1207 PR 1 replaces it with `parameter_list`. -pub(crate) fn default_size_params( - kind: SketchAlgorithm, - intent: &AggIntent, - eps: f64, - delta: f64, -) -> SketchParams { - crate::estimators::size_params(kind, intent, eps, delta) -} - #[cfg(test)] mod tests { use super::*; + use asap_types::ir::schema::{SketchAlgorithm, SketchParams}; /// CMS width is `⌈e/ε⌉` and depth `⌈ln(1/δ)⌉`, with an ε-only target /// taking `DEFAULT_DELTA` (depth 5). @@ -111,7 +95,7 @@ mod tests { accuracy: AccuracyTarget::max_error(0.001).failure_probability(0.001), }; assert_eq!( - default_size_params(SketchAlgorithm::Cms, &intent, 0.001, 0.001), + crate::estimators::size_params(SketchAlgorithm::Cms, &intent, 0.001, 0.001), SketchParams::Cms { width: 2719, depth: 7 @@ -122,14 +106,14 @@ mod tests { assert_eq!((eps, delta), (0.001, DEFAULT_DELTA)); let intent = AggIntent::Count { accuracy: target }; assert_eq!( - default_size_params(SketchAlgorithm::Cms, &intent, eps, delta), + crate::estimators::size_params(SketchAlgorithm::Cms, &intent, eps, delta), SketchParams::Cms { width: 2719, depth: 5 } ); assert_eq!( - default_size_params(SketchAlgorithm::Cms, &intent, 0.01, 0.01), + crate::estimators::size_params(SketchAlgorithm::Cms, &intent, 0.01, 0.01), SketchParams::Cms { width: 272, depth: 5 diff --git a/crates/accuracy/src/test_support.rs b/crates/accuracy/src/test_support.rs index 9a206a27f..c92e9ff58 100644 --- a/crates/accuracy/src/test_support.rs +++ b/crates/accuracy/src/test_support.rs @@ -5,12 +5,9 @@ use std::rc::Rc; use asap_types::ir::operator::operator_properties::Reduction; -use asap_types::ir::operator::AggIntent; use asap_types::ir::properties::guarantee::AccuracyGuarantee; use asap_types::ir::properties::AccuracyTarget; -use asap_types::ir::schema::{ - FieldDataType, SketchAlgorithm, SketchParams, SketchStatistic, SummaryUpdate, -}; +use asap_types::ir::schema::{FieldDataType, SketchStatistic, SummaryUpdate}; use asap_types::ir::{OperatorNode, ScalarExpr}; use crate::composition::StreamNorm; @@ -85,11 +82,6 @@ pub fn local_guarantee( crate::estimators::local_guarantee(family, query) } -/// `crate::estimators::cms::cms_width`. -pub fn cms_width(eps: f64) -> u32 { - crate::estimators::cms::cms_width(eps) -} - /// `crate::estimators::cms::cms_depth`. pub fn cms_depth(delta: f64) -> u32 { crate::estimators::cms::cms_depth(delta) @@ -100,16 +92,6 @@ pub fn kll_rank_error_99(k: u32) -> f64 { crate::estimators::kll::kll_rank_error_99(k) } -/// `crate::sizing::default_size_params`. -pub fn default_size_params( - kind: SketchAlgorithm, - intent: &AggIntent, - eps: f64, - delta: f64, -) -> SketchParams { - crate::sizing::default_size_params(kind, intent, eps, delta) -} - /// `crate::sizing::DEFAULT_DELTA`. pub const DEFAULT_DELTA: f64 = crate::sizing::DEFAULT_DELTA; diff --git a/crates/types/src/physical/properties.rs b/crates/types/src/physical/properties.rs index 3bc12bf0a..03a8effdd 100644 --- a/crates/types/src/physical/properties.rs +++ b/crates/types/src/physical/properties.rs @@ -64,8 +64,12 @@ impl PhysicalProperties { timing: ExecutionTiming, lifetime: Lifetime, ) -> Result { - let _ = (timing, lifetime); - todo!("#1207 PR 1: reject the invalid combination of section 2") + match (timing, lifetime) { + (ExecutionTiming::IngestionTime, Lifetime::OneEvaluation(_)) => { + Err(PhysicalPropertiesError::OneEvaluationAtIngestion) + } + _ => Ok(Self { timing, lifetime }), + } } /// $\tau_v$. @@ -89,32 +93,35 @@ mod tests { use super::*; use ExecutionTiming::{IngestionTime, QueryTime}; - const MEMORY: StorageMedium = StorageMedium::Memory; - - /// The one invalid combination of the concepts doc, section 2. - #[test] - fn new_rejects_one_evaluation_at_ingestion() { - assert_eq!( - PhysicalProperties::new(IngestionTime, Lifetime::OneEvaluation(MEMORY)), - Err(PhysicalPropertiesError::OneEvaluationAtIngestion) - ); - } - - /// The valid (τ, ℓ) pairs of section 2. + /// `new` accepts every (τ, ℓ) except (IngestionTime, OneEvaluation), for + /// both media, and keeps what it was given. #[test] - fn new_accepts_each_valid_pair() { - for (timing, lifetime) in [ - (QueryTime, Lifetime::None), - (IngestionTime, Lifetime::None), - (QueryTime, Lifetime::OneEvaluation(MEMORY)), - (QueryTime, Lifetime::AcrossEvaluations(MEMORY)), - ( - IngestionTime, - Lifetime::AcrossEvaluations(StorageMedium::Disk), - ), - ] { - let properties = PhysicalProperties::new(timing, lifetime).unwrap(); - assert_eq!(properties.is_materialized(), lifetime != Lifetime::None); + fn new_over_every_timing_and_lifetime() { + for medium in [StorageMedium::Memory, StorageMedium::Disk] { + for (timing, lifetime, valid) in [ + (QueryTime, Lifetime::None, true), + (QueryTime, Lifetime::OneEvaluation(medium), true), + (QueryTime, Lifetime::AcrossEvaluations(medium), true), + (IngestionTime, Lifetime::None, true), + (IngestionTime, Lifetime::OneEvaluation(medium), false), + (IngestionTime, Lifetime::AcrossEvaluations(medium), true), + ] { + let properties = PhysicalProperties::new(timing, lifetime); + match valid { + true => { + let properties = properties.unwrap(); + assert_eq!( + (properties.timing(), properties.lifetime()), + (timing, lifetime) + ); + assert_eq!(properties.is_materialized(), lifetime != Lifetime::None); + } + false => assert_eq!( + properties, + Err(PhysicalPropertiesError::OneEvaluationAtIngestion) + ), + } + } } } } diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index 76762609a..0d5ac6100 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -8,7 +8,10 @@ use std::collections::HashSet; use super::{GroupProperties, MemoGroup, MemoGroupAsInput, MemoGroupId}; use crate::ir::properties::summary_coverage::SummaryCoverage; use crate::ir::properties::AccuracyMetric; -use crate::ir::schema::{SketchAlgorithm, SketchParams}; +use crate::ir::schema::{ + FamilyCapability, FieldDataType, GroupingStrategy, HydraKind, HydraParams, SketchAlgorithm, + SketchParams, +}; use crate::ir::ASAPOp; use crate::ir::Operator; use crate::ir::QueryRoot; @@ -76,8 +79,7 @@ impl Memo { /// The candidate `id`. pub fn candidate(&self, id: CandidateForMemoGroupId) -> &CandidateForMemoGroup { - let _ = id; - todo!("#1207 PR 2") + &self.groups[id.group.0 as usize].candidates[id.index_in_group] } /// Replaces candidate `id`'s input requirements $r^{\log}_{c,i}$, one @@ -245,8 +247,20 @@ impl LogicalProperties { expression: &CandidateExpression, parameters: Option, ) -> Result { - let _ = (expression, parameters); - todo!("#1207 PR 1: reject invalid parameters") + match (family_of_operator(&expression.operator), ¶meters) { + (Some(_), None) => return Err(LogicalPropertiesError::SketchBuildWithoutParameters), + (None, Some(_)) => return Err(LogicalPropertiesError::ParametersOnExactOperator), + (Some(family), Some(parameters)) => { + if family != family_of_parameters(parameters) { + return Err(LogicalPropertiesError::ParametersOfAnotherFamily); + } + if !parameter_list(family).contains(parameters) { + return Err(LogicalPropertiesError::ParametersNotInList); + } + } + (None, None) => {} + } + Ok(Self { parameters }) } /// $\boldsymbol\theta_v$. @@ -272,9 +286,305 @@ pub enum LogicalPropertiesError { /// The family's predefined $\boldsymbol\theta$ list (#1207 step 6), each with /// its error bound derived through `estimators::local_guarantee` for a given /// statistic. +/// +/// Each list is today's sizing (`asap_accuracy`'s `size_params`) at +/// failure probability 0.01 over the errors ε = 0.1, 0.05, 0.02, 0.01, +/// 0.005, 0.002 and 0.001, weakest first, without repeats; an +/// `asap_accuracy` test keeps them equal. Distinct counts (KMV, and +/// UnivMon's distinct-count shapes) stop at ε = 0.01. A UnivMon's list is +/// its L2-norm shapes, then its distinct-count shapes, then its shape for +/// every other statistic. A heap variant combines each width × depth with +/// each heap size of [`HEAP_SIZES`], by width, then heap size: a top-k +/// consumer's floor is heap_size ≥ k, and the next listed step up is at most +/// 2.5× too large. Empty for a family Pass 1 does not offer (the Hydra +/// families included). pub fn parameter_list(family: SketchAlgorithm) -> &'static [SketchParams] { - let _ = family; - todo!("#1207 PR 1: the predefined θ list per family") + match family { + SketchAlgorithm::Kll => &KLL_PARAMETERS, + SketchAlgorithm::DDSketch => &DDSKETCH_PARAMETERS, + SketchAlgorithm::Hll => &HLL_PARAMETERS, + SketchAlgorithm::Kmv => &KMV_PARAMETERS, + SketchAlgorithm::Cms => &CMS_PARAMETERS, + SketchAlgorithm::CountSketch => &COUNT_SKETCH_PARAMETERS, + SketchAlgorithm::CmsWithHeap => &CMS_WITH_HEAP_PARAMETERS, + SketchAlgorithm::CountSketchWithHeap => &COUNT_SKETCH_WITH_HEAP_PARAMETERS, + SketchAlgorithm::UnivMon => &UNIVMON_PARAMETERS, + // Listed, not `_`: a family `asap_sketchlib` adds fails to compile + // here until it gets a list or is listed as not offered. + SketchAlgorithm::CountSketchWithL2Norm + | SketchAlgorithm::CmsWithSpaceSaving + | SketchAlgorithm::CountSketchWithSpaceSaving + | SketchAlgorithm::CmsFold + | SketchAlgorithm::CountSketchFold + | SketchAlgorithm::HllHip + | SketchAlgorithm::KllDynamic + | SketchAlgorithm::SpaceSaving + | SketchAlgorithm::Elastic + | SketchAlgorithm::Coco + | SketchAlgorithm::Bloom + | SketchAlgorithm::SetAggregator + | SketchAlgorithm::UnivMonTwoTier + | SketchAlgorithm::UnivMonQ + | SketchAlgorithm::HydraCms + | SketchAlgorithm::HydraCountSketch + | SketchAlgorithm::HydraKll + | SketchAlgorithm::HydraHll + | SketchAlgorithm::HydraUnivMon => &[], + } +} + +const KLL_PARAMETERS: [SketchParams; 7] = [ + kll(26), + kll(52), + kll(132), + kll(269), + kll(547), + kll(1404), + kll(2863), +]; + +const DDSKETCH_PARAMETERS: [SketchParams; 7] = [ + SketchParams::DDSketch { alpha: 0.1 }, + SketchParams::DDSketch { alpha: 0.05 }, + SketchParams::DDSketch { alpha: 0.02 }, + SketchParams::DDSketch { alpha: 0.01 }, + SketchParams::DDSketch { alpha: 0.005 }, + SketchParams::DDSketch { alpha: 0.002 }, + SketchParams::DDSketch { alpha: 0.001 }, +]; + +/// HLL's precision saturates at 18 from ε = 0.01 on. +const HLL_PARAMETERS: [SketchParams; 4] = + [hll(13, 0.1), hll(15, 0.05), hll(17, 0.02), hll(18, 0.01)]; + +const KMV_PARAMETERS: [SketchParams; 4] = [ + kmv(1_118, 0.1), + kmv(4_323, 0.05), + kmv(26_656, 0.02), + kmv(106_263, 0.01), +]; + +const CMS_PARAMETERS: [SketchParams; 7] = [ + SketchParams::Cms { + width: 28, + depth: 5, + }, + SketchParams::Cms { + width: 55, + depth: 5, + }, + SketchParams::Cms { + width: 136, + depth: 5, + }, + SketchParams::Cms { + width: 272, + depth: 5, + }, + SketchParams::Cms { + width: 544, + depth: 5, + }, + SketchParams::Cms { + width: 1_360, + depth: 5, + }, + SketchParams::Cms { + width: 2_719, + depth: 5, + }, +]; + +const COUNT_SKETCH_PARAMETERS: [SketchParams; 7] = [ + SketchParams::CountSketch { + width: 300, + depth: 47, + }, + SketchParams::CountSketch { + width: 1_200, + depth: 47, + }, + SketchParams::CountSketch { + width: 7_500, + depth: 47, + }, + SketchParams::CountSketch { + width: 30_000, + depth: 47, + }, + SketchParams::CountSketch { + width: 120_000, + depth: 47, + }, + SketchParams::CountSketch { + width: 750_000, + depth: 47, + }, + SketchParams::CountSketch { + width: 3_000_000, + depth: 47, + }, +]; + +/// The heap sizes of a heap variant's list: the 1-2-5 series from 10 to +/// 10 000, so a top-k with k ≤ 10 000 has a listed heap. +const HEAP_SIZES: [u32; 10] = [10, 20, 50, 100, 200, 500, 1_000, 2_000, 5_000, 10_000]; + +const CMS_WITH_HEAP_PARAMETERS: [SketchParams; 70] = with_heap_sizes(&CMS_PARAMETERS); + +const COUNT_SKETCH_WITH_HEAP_PARAMETERS: [SketchParams; 70] = + with_heap_sizes(&COUNT_SKETCH_PARAMETERS); + +/// Each `Cms` (or `CountSketch`) shape of `shapes` with each heap size of +/// [`HEAP_SIZES`], as `CmsWithHeap` (or `CountSketchWithHeap`). +const fn with_heap_sizes(shapes: &[SketchParams; 7]) -> [SketchParams; 70] { + const UNSET: SketchParams = SketchParams::Cms { width: 0, depth: 0 }; + let mut list = [UNSET; 70]; + let mut shape = 0; + while shape < shapes.len() { + let mut heap = 0; + while heap < HEAP_SIZES.len() { + let heap_size = HEAP_SIZES[heap]; + list[shape * HEAP_SIZES.len() + heap] = match shapes[shape] { + SketchParams::Cms { width, depth } => SketchParams::CmsWithHeap { + width, + depth, + heap_size, + }, + SketchParams::CountSketch { width, depth } => SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }, + _ => panic!("a heap variant extends a Cms or CountSketch shape"), + }; + heap += 1; + } + shape += 1; + } + list +} + +const UNIVMON_PARAMETERS: [SketchParams; 12] = [ + univmon_l2(624), + univmon_l2(2_367), + univmon_l2(14_349), + univmon_l2(56_817), + univmon_l2(226_130), + univmon_l2(1_409_067), + univmon_l2(5_630_630), + univmon_distinct(13_500), + univmon_distinct(54_000), + univmon_distinct(337_500), + univmon_distinct(1_350_000), + SketchParams::UnivMon { + heap_size: 256, + sketch_rows: 5, + sketch_cols: 1_024, + layers: 16, + distinct_failure_probability: None, + }, +]; + +const fn kll(k: u32) -> SketchParams { + SketchParams::Kll { + k, + failure_probability: None, + } +} + +const fn hll(precision: u8, relative_error: f64) -> SketchParams { + SketchParams::Hll { + precision, + relative_error: Some(relative_error), + } +} + +const fn kmv(k: u32, relative_error: f64) -> SketchParams { + SketchParams::Kmv { + k, + relative_error: Some(relative_error), + } +} + +const fn univmon_l2(sketch_cols: u32) -> SketchParams { + SketchParams::UnivMon { + heap_size: 1, + sketch_rows: 5, + sketch_cols, + layers: 1, + distinct_failure_probability: None, + } +} + +const fn univmon_distinct(heap_size: u32) -> SketchParams { + SketchParams::UnivMon { + heap_size, + sketch_rows: 1, + sketch_cols: 1, + layers: 19, + distinct_failure_probability: Some(0.01), + } +} + +/// The sketch family an operator builds: a `SummaryAgg`'s or a +/// `SummaryTumblingWindowChain`'s sketch family (a Hydra one's Hydra +/// family); `None` for every other operator, an exact accumulator build +/// included. +fn family_of_operator(operator: &Operator) -> Option { + let (Operator::ASAP(ASAPOp::SummaryAgg { family, .. }) + | Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { family, .. })) = operator + else { + return None; + }; + match family { + FieldDataType::Sketch(algorithm, GroupingStrategy::PerSubpopulationInstance) => { + Some(*algorithm) + } + FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind }) => { + Some(match kind { + HydraKind::HydraKll => SketchAlgorithm::HydraKll, + HydraKind::HydraCms => SketchAlgorithm::HydraCms, + HydraKind::HydraCountSketch => SketchAlgorithm::HydraCountSketch, + }) + } + FieldDataType::Plain(_) + | FieldDataType::ExactAggregate(..) + | FieldDataType::SummarySample(..) + | FieldDataType::Wavelet(..) + | FieldDataType::StatModel(..) => None, + } +} + +/// The family `parameters` size. +fn family_of_parameters(parameters: &SketchParams) -> SketchAlgorithm { + match parameters { + SketchParams::UnivMon { .. } => SketchAlgorithm::UnivMon, + SketchParams::Kll { .. } => SketchAlgorithm::Kll, + SketchParams::Cms { .. } => SketchAlgorithm::Cms, + SketchParams::Hll { .. } => SketchAlgorithm::Hll, + SketchParams::DDSketch { .. } => SketchAlgorithm::DDSketch, + SketchParams::CmsWithHeap { .. } => SketchAlgorithm::CmsWithHeap, + SketchParams::Kmv { .. } => SketchAlgorithm::Kmv, + SketchParams::CountSketch { .. } => SketchAlgorithm::CountSketch, + SketchParams::CountSketchWithHeap { .. } => SketchAlgorithm::CountSketchWithHeap, + SketchParams::Hydra(HydraParams::HydraKll { .. }) => SketchAlgorithm::HydraKll, + SketchParams::Hydra(HydraParams::HydraCms { .. }) => SketchAlgorithm::HydraCms, + SketchParams::Hydra(HydraParams::HydraCountSketch { .. }) => { + SketchAlgorithm::HydraCountSketch + } + } +} + +/// Whether a tumbling window layout fits the operator: a `SummaryAgg` whose +/// states merge, so that its windows' states make the whole range's. +fn can_be_windowed(operator: &Operator) -> bool { + match operator { + Operator::ASAP(ASAPOp::SummaryAgg { family, .. }) => { + FamilyCapability::of(family).merge_inputs.is_some() + } + Operator::ASAP(_) | Operator::NonASAP(_) => false, + } } /// $r^{\log}_{c,i}$: only what $G_{c,i}$'s identity does not fix. @@ -334,6 +644,515 @@ pub fn producer_satisfies_logical_requirements( producer: &LogicalProperties, requirement: &LogicalRequirements, ) -> bool { - let _ = (producer, requirement); - todo!("#1207 PR 1: per-family dominance") + match ( + producer.parameters(), + requirement.parameter_floor_for_required_accuracy(), + ) { + (_, None) | (None, Some(_)) => true, + (Some(parameters), Some(floor)) => dominates(parameters, floor), + } +} + +/// $\boldsymbol\theta \succeq \boldsymbol\theta^{\min}$: every parameter +/// that sets $\varepsilon$ is at least the floor's (at most, for DDSketch's +/// `alpha`). The recorded failure probabilities and relative errors are not +/// compared: every listed configuration of a family records the same +/// failure probability, 0.01. Parameters of different families never +/// dominate. +fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { + use SketchParams as P; + match (parameters, floor) { + (P::Kll { k, .. }, P::Kll { k: floor_k, .. }) + | (P::Kmv { k, .. }, P::Kmv { k: floor_k, .. }) => k >= floor_k, + (P::DDSketch { alpha }, P::DDSketch { alpha: floor_alpha }) => alpha <= floor_alpha, + ( + P::Hll { precision, .. }, + P::Hll { + precision: floor_precision, + .. + }, + ) => precision >= floor_precision, + (P::Cms { width, depth }, P::Cms { width: w, depth: d }) + | (P::CountSketch { width, depth }, P::CountSketch { width: w, depth: d }) => { + width >= w && depth >= d + } + ( + P::CmsWithHeap { + width, + depth, + heap_size, + }, + P::CmsWithHeap { + width: w, + depth: d, + heap_size: h, + }, + ) + | ( + P::CountSketchWithHeap { + width, + depth, + heap_size, + }, + P::CountSketchWithHeap { + width: w, + depth: d, + heap_size: h, + }, + ) => width >= w && depth >= d && heap_size >= h, + ( + P::UnivMon { + heap_size, + sketch_rows, + sketch_cols, + layers, + .. + }, + P::UnivMon { + heap_size: h, + sketch_rows: r, + sketch_cols: c, + layers: l, + .. + }, + ) => heap_size >= h && sketch_rows >= r && sketch_cols >= c && layers >= l, + ( + P::Hydra(HydraParams::HydraKll { k, shared_buckets }), + P::Hydra(HydraParams::HydraKll { + k: floor_k, + shared_buckets: b, + }), + ) => k >= floor_k && shared_buckets >= b, + ( + P::Hydra(HydraParams::HydraCms { + width, + depth, + shared_rows, + shared_columns, + }), + P::Hydra(HydraParams::HydraCms { + width: w, + depth: d, + shared_rows: r, + shared_columns: c, + }), + ) + | ( + P::Hydra(HydraParams::HydraCountSketch { + width, + depth, + shared_rows, + shared_columns, + }), + P::Hydra(HydraParams::HydraCountSketch { + width: w, + depth: d, + shared_rows: r, + shared_columns: c, + }), + ) => width >= w && depth >= d && shared_rows >= r && shared_columns >= c, + // Different families. Listed by `parameters`' variant, not `_`, so a + // new variant fails to compile here until it is compared above. + (P::Kll { .. }, _) + | (P::Kmv { .. }, _) + | (P::DDSketch { .. }, _) + | (P::Hll { .. }, _) + | (P::Cms { .. }, _) + | (P::CountSketch { .. }, _) + | (P::CmsWithHeap { .. }, _) + | (P::CountSketchWithHeap { .. }, _) + | (P::UnivMon { .. }, _) + | (P::Hydra(_), _) => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::ir::operator::operator_properties::Reduction; + use crate::ir::scalar::expr_ir::ColumnRef; + use crate::ir::schema::{ExactKind, SummaryUpdate}; + + /// `SummaryAgg` over group 0 building a state of `family`. + fn summary_agg(family: FieldDataType) -> CandidateExpression { + CandidateExpression { + operator: Operator::ASAP(ASAPOp::SummaryAgg { + child: MemoGroupId(0), + family, + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }), + } + } + + fn sketch(algorithm: SketchAlgorithm, grouping: GroupingStrategy) -> FieldDataType { + FieldDataType::Sketch(algorithm, grouping) + } + + const KLL_269: SketchParams = kll(269); + const CMS: SketchParams = SketchParams::Cms { + width: 272, + depth: 5, + }; + const HYDRA_CMS: SketchParams = SketchParams::Hydra(HydraParams::HydraCms { + width: 272, + depth: 5, + shared_rows: 3, + shared_columns: 1_024, + }); + + /// `LogicalProperties::new` over every operator kind, parameters and + /// window layout: a sketch build needs parameters of its family from its + /// list; parameters are rejected on an operator without a sketch family; + /// a window is rejected at width 0 or on an operator other than a + /// `SummaryAgg` whose states merge. Parameters are checked first. + #[test] + fn logical_properties_new_over_every_operator_parameters_and_window() { + use LogicalPropertiesError as Invalid; + let per_instance = GroupingStrategy::PerSubpopulationInstance; + let hydra = GroupingStrategy::SharedMultiSubpopulation { + kind: HydraKind::HydraCms, + }; + let kll_chain = CandidateExpression { + operator: Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: MemoGroupId(0), + family: sketch(SketchAlgorithm::Kll, per_instance.clone()), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: per_instance.clone(), + filter: None, + tumbling_window_width_ms: 5_000, + tumbling_window_count: 12, + }), + }; + let readout = CandidateExpression { + operator: Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: MemoGroupId(0), + query: crate::ir::schema::SketchStatistic::Quantile { q: 0.5 }, + }), + }; + // KLL_270 is of KLL's type but not in its list. + let parameters = [ + None, + Some(KLL_269), + Some(CMS), + Some(HYDRA_CMS), + Some(kll(270)), + ]; + let windows = [ + None, + Some(WindowLayout::Tumbling { width_ms: 5_000 }), + Some(WindowLayout::Tumbling { width_ms: 0 }), + ]; + let no_parameters = Err(Invalid::ParametersOnExactOperator); + let needs_parameters = Err(Invalid::SketchBuildWithoutParameters); + let not_in_list = Err(Invalid::ParametersNotInList); + let another_family = Err(Invalid::ParametersOfAnotherFamily); + let cannot_be_windowed = Err(Invalid::WindowOnOperatorThatCannotBeWindowed); + let zero_width = Err(Invalid::ZeroWindowWidth); + // Per operator: the result for each of `parameters`, then for each of + // `windows`; a combination fails with the parameters' error first. + let table = [ + ( + summary_agg(sketch(SketchAlgorithm::Kll, per_instance.clone())), + [ + needs_parameters, + Ok(()), + another_family, + another_family, + not_in_list, + ], + [Ok(()), Ok(()), zero_width], + ), + ( + // No Hydra θ is listed yet, so no Hydra build is valid. + summary_agg(sketch(SketchAlgorithm::Cms, hydra)), + [ + needs_parameters, + another_family, + another_family, + not_in_list, + another_family, + ], + [Ok(()), Ok(()), zero_width], + ), + ( + summary_agg(FieldDataType::exact(ExactKind::Sum)), + [ + Ok(()), + no_parameters, + no_parameters, + no_parameters, + no_parameters, + ], + [Ok(()), Ok(()), zero_width], + ), + // IRate's states do not merge across a window boundary. + ( + summary_agg(FieldDataType::exact(ExactKind::IRate)), + [ + Ok(()), + no_parameters, + no_parameters, + no_parameters, + no_parameters, + ], + [Ok(()), cannot_be_windowed, zero_width], + ), + ( + kll_chain, + [ + needs_parameters, + Ok(()), + another_family, + another_family, + not_in_list, + ], + [Ok(()), cannot_be_windowed, zero_width], + ), + ( + readout, + [ + Ok(()), + no_parameters, + no_parameters, + no_parameters, + no_parameters, + ], + [Ok(()), cannot_be_windowed, zero_width], + ), + ]; + for (expression, by_parameters, by_window) in table { + for (parameters, parameters_result) in parameters.iter().zip(by_parameters) { + for (window, window_result) in windows.iter().zip(by_window) { + let expected = parameters_result.and(window_result); + let properties = + LogicalProperties::new(&expression, parameters.clone(), *window); + assert_eq!( + properties.as_ref().map(|_| ()).map_err(|e| *e), + expected, + "{:?} {parameters:?} {window:?}", + expression.operator.kind_name() + ); + if let Ok(properties) = properties { + assert_eq!(properties.parameters(), parameters.as_ref()); + assert_eq!(properties.window_layout(), window.as_ref()); + } + } + } + } + } + + /// Each predefined list holds only its family's parameters, without + /// repeats, and is empty exactly for the families Pass 1 does not offer. + #[test] + fn parameter_list_holds_its_family_without_repeats() { + let offered = [ + SketchAlgorithm::Kll, + SketchAlgorithm::DDSketch, + SketchAlgorithm::Hll, + SketchAlgorithm::Kmv, + SketchAlgorithm::Cms, + SketchAlgorithm::CountSketch, + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + SketchAlgorithm::UnivMon, + ]; + for family in offered.iter().copied().chain([ + SketchAlgorithm::HydraCms, + SketchAlgorithm::HydraCountSketch, + SketchAlgorithm::HydraKll, + SketchAlgorithm::HllHip, + ]) { + let list = parameter_list(family); + assert_eq!(list.is_empty(), !offered.contains(&family), "{family:?}"); + for (i, parameters) in list.iter().enumerate() { + assert_eq!(family_of_parameters(parameters), family); + assert!(!list[..i].contains(parameters), "{parameters:?} repeats"); + } + } + } + + /// `LogicalRequirements::new` accepts no floor and any value of a family's + /// parameter type, listed or not (e.g. heap_size ≥ k exactly). + #[test] + fn logical_requirement_new_accepts_any_floor() { + assert_eq!( + LogicalRequirements::new(None).parameter_floor_for_required_accuracy(), + None + ); + for floor in [ + KLL_269, + kll(270), + SketchParams::DDSketch { alpha: 0.03 }, + SketchParams::CmsWithHeap { + width: 272, + depth: 5, + heap_size: 37, + }, + HYDRA_CMS, + ] { + let requirement = LogicalRequirements::new(Some(floor.clone())); + assert_eq!( + requirement.parameter_floor_for_required_accuracy(), + Some(&floor) + ); + } + } + + /// `producer_satisfies_logical_requirements`: no floor or an exact producer always satisfies; + /// otherwise every parameter that sets ε is at least the floor's (at + /// most, for DDSketch's `alpha`), and another family never satisfies. + #[test] + fn producer_satisfies_logical_requirements_compares_the_parameters_that_set_the_error() { + let list = |family| parameter_list(family); + let [kll_26, _, _, kll_269, kll_547, ..] = list(SketchAlgorithm::Kll) else { + unreachable!() + }; + let [dd_10, _, _, dd_1, ..] = list(SketchAlgorithm::DDSketch) else { + unreachable!() + }; + let [hll_13, _, _, hll_18] = list(SketchAlgorithm::Hll) else { + unreachable!() + }; + let [kmv_1118, kmv_4323, ..] = list(SketchAlgorithm::Kmv) else { + unreachable!() + }; + let [cms_28, cms_55, ..] = list(SketchAlgorithm::Cms) else { + unreachable!() + }; + let [cs_300, cs_1200, ..] = list(SketchAlgorithm::CountSketch) else { + unreachable!() + }; + let [cms_heap_10, cms_heap_20, ..] = list(SketchAlgorithm::CmsWithHeap) else { + unreachable!() + }; + let [cs_heap_10, cs_heap_20, ..] = list(SketchAlgorithm::CountSketchWithHeap) else { + unreachable!() + }; + let univmon = list(SketchAlgorithm::UnivMon); + let (l2_624, l2_2367, distinct_13500, univmon_other) = + (&univmon[0], &univmon[1], &univmon[7], &univmon[11]); + let build = |parameters: &SketchParams| { + let family = family_of_parameters(parameters); + LogicalProperties::new( + &summary_agg(sketch(family, GroupingStrategy::default())), + Some(parameters.clone()), + None, + ) + .unwrap() + }; + let floor = |parameters: &SketchParams| LogicalRequirements::new(Some(parameters.clone())); + let exact = LogicalProperties::new( + &summary_agg(FieldDataType::exact(ExactKind::Sum)), + None, + None, + ) + .unwrap(); + let no_floor = LogicalRequirements::new(None); + assert!(producer_satisfies_logical_requirements( + &build(kll_26), + &no_floor + )); + assert!(producer_satisfies_logical_requirements(&exact, &no_floor)); + assert!(producer_satisfies_logical_requirements( + &exact, + &floor(kll_547) + )); + // A top-37 needs heap_size ≥ 37: the next listed step, 50, satisfies. + let top_37 = SketchParams::CmsWithHeap { + width: 28, + depth: 5, + heap_size: 37, + }; + let heap = |heap_size| SketchParams::CmsWithHeap { + width: 28, + depth: 5, + heap_size, + }; + assert!(producer_satisfies_logical_requirements( + &build(&heap(50)), + &floor(&top_37) + )); + assert!(!producer_satisfies_logical_requirements( + &build(&heap(20)), + &floor(&top_37) + )); + for (producer, required, satisfies) in [ + (kll_547, kll_269, true), + (kll_269, kll_269, true), + (kll_269, kll_547, false), + (dd_1, dd_10, true), + (dd_10, dd_1, false), + (hll_18, hll_13, true), + (hll_13, hll_18, false), + (kmv_4323, kmv_1118, true), + (kmv_1118, kmv_4323, false), + (cms_55, cms_28, true), + (cms_28, cms_55, false), + (cs_1200, cs_300, true), + (cs_300, cs_1200, false), + (cms_heap_20, cms_heap_10, true), + (cms_heap_10, cms_heap_20, false), + (cs_heap_20, cs_heap_10, true), + (cs_heap_10, cs_heap_20, false), + (l2_2367, l2_624, true), + (l2_624, l2_2367, false), + // A wider, deeper and heavier UnivMon dominates each L2 shape it + // covers, but no shape dominates one with a larger heap. + (univmon_other, l2_624, true), + (univmon_other, l2_2367, false), + (univmon_other, distinct_13500, false), + (distinct_13500, univmon_other, false), + // Another family. + (kll_547, dd_10, false), + (cms_55, cs_300, false), + (cms_heap_20, cms_28, false), + ] { + assert_eq!( + producer_satisfies_logical_requirements(&build(producer), &floor(required)), + satisfies, + "{producer:?} ⪰ {required:?}" + ); + } + } + + /// Hydra configurations dominate per kind on every size, and never across + /// kinds (no Hydra floor is listed yet, so this checks `dominates` alone). + #[test] + fn hydra_dominance_compares_every_size_of_one_kind() { + let hydra_cms = |width, shared_columns| { + SketchParams::Hydra(HydraParams::HydraCms { + width, + depth: 5, + shared_rows: 3, + shared_columns, + }) + }; + let hydra_count_sketch = SketchParams::Hydra(HydraParams::HydraCountSketch { + width: 300, + depth: 47, + shared_rows: 3, + shared_columns: 1_024, + }); + let hydra_kll = + |k, shared_buckets| SketchParams::Hydra(HydraParams::HydraKll { k, shared_buckets }); + for (parameters, floor, dominates_floor) in [ + (hydra_cms(272, 1_024), hydra_cms(272, 1_024), true), + (hydra_cms(544, 2_048), hydra_cms(272, 1_024), true), + (hydra_cms(544, 512), hydra_cms(272, 1_024), false), + (hydra_kll(269, 64), hydra_kll(132, 64), true), + (hydra_kll(269, 32), hydra_kll(132, 64), false), + (hydra_count_sketch.clone(), hydra_cms(28, 16), false), + (hydra_cms(28, 16), CMS, false), + ] { + assert_eq!( + dominates(¶meters, &floor), + dominates_floor, + "{parameters:?} ⪰ {floor:?}" + ); + } + } } diff --git a/crates/types/src/stages/physical/memo_candidates.rs b/crates/types/src/stages/physical/memo_candidates.rs index 7070ef3f3..c1807805e 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -9,8 +9,11 @@ use std::collections::HashMap; use super::execution_context::ExecutionContext; use super::graph::PhysicalCandidateId; use crate::ir::QueryRoot; -use crate::physical::PhysicalProperties; -use crate::stages::{CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, PrunedCandidate}; +use crate::physical::{ExecutionTiming, Lifetime, PhysicalProperties}; +use crate::stages::{ + CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, MemoGroupId, PrunedCandidate, +}; +use crate::workload::{QueryRecurrence, RepeatedDemand}; /// Index into [`PhysicalASAPMemoCandidates::edges`]. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] @@ -35,9 +38,39 @@ impl PhysicalCandidate { /// and phase), each scheduled time, or a one-off query's time; false /// when the times are not known (an interval without a phase, an /// estimated rate, an unknown recurrence). + /// + /// A `None` or `OneEvaluation` producer runs in the consumer's own + /// context ($x_{\kappa,v}$, Stage 3), so its result belongs to that + /// context. A non-windowed `AcrossEvaluations` producer is not offered + /// yet (it needs its own delivery rule), so this returns false for it. pub fn delivers_on(&self, memo: &Memo, context: &ExecutionContext) -> bool { - let _ = (memo, context); - todo!("#1207 PR 1: the alignment table of section 2") + match self.physical_properties.lifetime() { + Lifetime::None | Lifetime::OneEvaluation(_) => true, + Lifetime::AcrossEvaluations(_) => { + let Some(WindowLayout::Tumbling { width_ms }) = memo + .candidate(self.logical_candidate_id) + .logical_properties + .window_layout() + else { + return false; + }; + match context { + ExecutionContext::Evaluation { + recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(_)), + .. + } => false, + ExecutionContext::Evaluation { + recurrence, + range_evaluation, + } => recurrence + .evaluation_grid_ms(*range_evaluation) + .is_some_and(|grid_ms| grid_ms.is_multiple_of(*width_ms)), + ExecutionContext::IngestionPipeline(_) | ExecutionContext::Materialized(_) => { + todo!("#1207 PR 7: delivery into an ingestion or materialized context") + } + } + } + } } } @@ -48,8 +81,20 @@ pub fn producer_satisfies_physical_requirements( producer: &PhysicalCandidate, consumer: &PhysicalCandidate, ) -> bool { - let _ = (producer, consumer); - todo!("#1207 PR 1: the timing x lifetime table of section 2") + use ExecutionTiming::{IngestionTime, QueryTime}; + let consumer_timing = consumer.physical_properties.timing(); + match ( + producer.physical_properties.timing(), + producer.physical_properties.lifetime(), + ) { + (QueryTime, _) => consumer_timing == QueryTime, + (IngestionTime, Lifetime::None) => consumer_timing == IngestionTime, + // Materialized at ingestion: read at ingestion time and at query time. + (IngestionTime, Lifetime::AcrossEvaluations(_)) => true, + (IngestionTime, Lifetime::OneEvaluation(_)) => { + unreachable!("rejected by `PhysicalProperties::new`") + } + } } /// $(u, c, i)$ over physical candidates. @@ -152,3 +197,271 @@ pub fn output_references(query_roots: &[QueryRoot]) -> Vec<(usize, } references } + +#[cfg(test)] +mod tests { + use std::rc::Rc; + + use super::*; + use crate::ir::operator::operator_properties::Reduction; + use crate::ir::scalar::expr_ir::{ColumnRef, ScalarValue}; + use crate::ir::schema::{ + DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchParams, + SummaryUpdate, + }; + use crate::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, ScalarExpr}; + use crate::physical::StorageMedium; + use crate::stages::{ + CandidateExpression, CandidateForMemoGroup, LogicalProperties, MemoGroup, + OriginOfCandidateForMemoGroup, + }; + use crate::workload::{DemandEstimate, RangeEvaluation, Rate, RepetitionInterval, TimestampMs}; + use ExecutionTiming::{IngestionTime, QueryTime}; + + /// Candidate 0 of the test memo has no window layout. + const UNWINDOWED: CandidateForMemoGroupId = CandidateForMemoGroupId { + group: MemoGroupId(0), + index_in_group: 0, + }; + /// Candidate 1 has tumbling windows 5 s wide. + const WINDOWS_5S: CandidateForMemoGroupId = CandidateForMemoGroupId { + group: MemoGroupId(0), + index_in_group: 1, + }; + /// Candidate 2 has tumbling windows 10 s wide. + const WINDOWS_10S: CandidateForMemoGroupId = CandidateForMemoGroupId { + group: MemoGroupId(0), + index_in_group: 2, + }; + + /// One group of KLL states with three candidates: no window layout, + /// 5 s tumbling windows and 10 s tumbling windows. + fn memo() -> Memo { + let expression = CandidateExpression { + operator: Operator::ASAP(ASAPOp::SummaryAgg { + child: MemoGroupId(0), + family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }), + }; + let candidate = |window_layout| CandidateForMemoGroup { + logical_properties: LogicalProperties::new( + &expression, + Some(SketchParams::Kll { + k: 269, + failure_probability: None, + }), + window_layout, + ) + .unwrap(), + expression: expression.clone(), + input_requirements: vec![], + fused_groups: vec![], + origin: OriginOfCandidateForMemoGroup::Pass1, + }; + let values = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values { + rows: vec![vec![ScalarExpr::Literal(ScalarValue::Float64(1.0))]], + schema: Schema::lifted(vec![Field::plain("value", DataType::Float64, false)], None), + })) + .unwrap(); + Memo { + groups: vec![MemoGroup { + logical_node: Rc::clone(&values), + child_groups: vec![], + parents: vec![], + queries: vec![], + candidates: vec![ + candidate(None), + candidate(Some(WindowLayout::Tumbling { width_ms: 5_000 })), + candidate(Some(WindowLayout::Tumbling { width_ms: 10_000 })), + ], + }], + query_roots: vec![], + } + } + + fn candidate( + logical_candidate_id: CandidateForMemoGroupId, + timing: ExecutionTiming, + lifetime: Lifetime, + ) -> PhysicalCandidate { + PhysicalCandidate { + logical_candidate_id, + physical_properties: PhysicalProperties::new(timing, lifetime).unwrap(), + } + } + + fn evaluation( + recurrence: QueryRecurrence, + range_evaluation: Option, + ) -> ExecutionContext { + ExecutionContext::Evaluation { + recurrence, + range_evaluation, + } + } + + fn every(interval_ms: u32, phase_ms: u64) -> QueryRecurrence { + QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval_ms), + evaluation_phase: TimestampMs(phase_ms), + }) + } + + const MEMORY: StorageMedium = StorageMedium::Memory; + + /// `delivers_on` follows the alignment table, one case per recurrence: a + /// windowed `AcrossEvaluations` producer of width $w$ (the window + /// layout's, grid at phase 0) needs $w$ to divide the context's + /// evaluation grid (interval, range step and phase), each scheduled time, + /// or a one-off query's time, and is refused when the times are not + /// known; a `None` or `OneEvaluation` producer runs in the consumer's + /// context and delivers on any; a non-windowed `AcrossEvaluations` + /// producer is not offered. + #[test] + fn delivers_on_follows_the_alignment_table() { + let memo = memo(); + let across_5s = candidate(WINDOWS_5S, QueryTime, Lifetime::AcrossEvaluations(MEMORY)); + let ingestion_10s = candidate( + WINDOWS_10S, + IngestionTime, + Lifetime::AcrossEvaluations(MEMORY), + ); + let across_unwindowed = + candidate(UNWINDOWED, QueryTime, Lifetime::AcrossEvaluations(MEMORY)); + let unmaterialized = candidate(UNWINDOWED, QueryTime, Lifetime::None); + let one_evaluation = candidate(WINDOWS_5S, QueryTime, Lifetime::OneEvaluation(MEMORY)); + let hour_by_minute = Some(RangeEvaluation { + step_ms: 60_000, + span_ms: 3_600_000, + }); + let hour_by_7s = Some(RangeEvaluation { + step_ms: 7_000, + span_ms: 3_600_000, + }); + let scheduled = |times: &[u64]| { + QueryRecurrence::Repeated(RepeatedDemand::Scheduled( + times.iter().copied().map(TimestampMs).collect(), + )) + }; + let one_off = |execute_at| QueryRecurrence::OneTime { + invocations: 3, + execute_at, + }; + let estimated = QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(DemandEstimate { + expected_rate: Rate(0.1), + observed_at: None, + valid_for: None, + })); + let without_phase = + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(10_000))); + for (producer, context, delivers) in [ + // FixedIntervalAt: w divides gcd(interval, step, phase). + (across_5s, evaluation(every(10_000, 0), None), true), + (across_5s, evaluation(every(10_000, 5_000), None), true), + (across_5s, evaluation(every(10_000, 2_000), None), false), + (across_5s, evaluation(every(7_000, 0), None), false), + ( + across_5s, + evaluation(every(10_000, 0), hour_by_minute), + true, + ), + (across_5s, evaluation(every(10_000, 0), hour_by_7s), false), + // Phases 0 and 5 s with w = 10 s: only w = 5 s serves both. + (ingestion_10s, evaluation(every(10_000, 0), None), true), + (ingestion_10s, evaluation(every(10_000, 5_000), None), false), + // FixedInterval without a phase: the times are not known. + (across_5s, evaluation(without_phase.clone(), None), false), + // Scheduled: w divides every time. + ( + across_5s, + evaluation(scheduled(&[10_000, 25_000]), None), + true, + ), + ( + across_5s, + evaluation(scheduled(&[10_000, 27_000]), None), + false, + ), + // OneTime: w divides its time; without a time, not known. + ( + across_5s, + evaluation(one_off(Some(TimestampMs(60_000))), None), + true, + ), + ( + across_5s, + evaluation(one_off(Some(TimestampMs(61_000))), None), + false, + ), + (across_5s, evaluation(one_off(None), None), false), + // EstimatedRate and Unknown: not known. + (across_5s, evaluation(estimated.clone(), None), false), + (across_5s, evaluation(QueryRecurrence::Unknown, None), false), + (across_unwindowed, evaluation(every(10_000, 0), None), false), + ( + unmaterialized, + evaluation(without_phase.clone(), None), + true, + ), + (one_evaluation, evaluation(estimated.clone(), None), true), + ] { + assert_eq!( + producer.delivers_on(&memo, &context), + delivers, + "{:?} {:?} in {context:?}", + producer.logical_candidate_id, + producer.physical_properties + ); + } + } + + /// `producer_satisfies_physical_requirements` over every valid producer state and every valid + /// consumer state: a query-time producer serves only query-time + /// consumers, an ingestion-time producer that is not materialized only + /// ingestion-time consumers, and one materialized across evaluations + /// serves both. Window layouts do not matter: alignment is Stage 3's + /// per-context check. + #[test] + fn producer_satisfies_physical_requirements_over_every_valid_timing_and_lifetime() { + let states = |medium| { + [ + candidate(UNWINDOWED, QueryTime, Lifetime::None), + candidate(UNWINDOWED, QueryTime, Lifetime::OneEvaluation(medium)), + candidate(WINDOWS_10S, QueryTime, Lifetime::AcrossEvaluations(medium)), + candidate(UNWINDOWED, IngestionTime, Lifetime::None), + candidate( + WINDOWS_5S, + IngestionTime, + Lifetime::AcrossEvaluations(medium), + ), + ] + }; + // Rows: producer state; columns: consumer state, in `states` order. + let expected = [ + [true, true, true, false, false], + [true, true, true, false, false], + [true, true, true, false, false], + [false, false, false, true, true], + [true, true, true, true, true], + ]; + for producer_medium in [StorageMedium::Memory, StorageMedium::Disk] { + for consumer_medium in [StorageMedium::Memory, StorageMedium::Disk] { + for (producer, row) in states(producer_medium).iter().zip(expected) { + for (consumer, satisfies) in states(consumer_medium).iter().zip(row) { + assert_eq!( + producer_satisfies_physical_requirements(producer, consumer), + satisfies, + "{:?} -> {:?}", + producer.physical_properties, + consumer.physical_properties + ); + } + } + } + } + } +} diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs index 3558df41c..407ca58a3 100644 --- a/crates/types/src/workload/mod.rs +++ b/crates/types/src/workload/mod.rs @@ -209,6 +209,48 @@ impl QueryRecurrence { _ => None, } } + + /// The grid every evaluation with this recurrence and `range_evaluation` + /// falls on, in milliseconds, anchored at the epoch (every evaluation + /// time is a multiple of it). The times' grid is the fixed interval + /// ([`Self::fixed_interval_ms`]; with a declared + /// [`RepeatedDemand::FixedIntervalAt`] phase, the greatest common divisor + /// with that phase: every 20 s at phase 5 s is on a 5 s grid), the + /// greatest common divisor of the scheduled times, or a one-off query's + /// time. A range evaluation of more than one step also takes the greatest + /// common divisor with its step, since each run evaluates at every step + /// up to its end. `None` when the times are not known: an estimated + /// rate, a one-off query without a time, an unknown recurrence, or a + /// zero interval. A fixed interval without a declared phase is taken at + /// phase 0. + pub fn evaluation_grid_ms(&self, range_evaluation: Option) -> Option { + let times_grid_ms = match self { + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(_)) => { + self.fixed_interval_ms()? + } + QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + evaluation_phase, .. + }) => greatest_common_divisor(self.fixed_interval_ms()?, evaluation_phase.0), + QueryRecurrence::Repeated(RepeatedDemand::Scheduled(times)) => times + .iter() + .fold(0, |grid_ms, time| greatest_common_divisor(grid_ms, time.0)), + QueryRecurrence::OneTime { + execute_at: Some(time), + .. + } => time.0, + QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(_)) + | QueryRecurrence::OneTime { + execute_at: None, .. + } + | QueryRecurrence::Unknown => return None, + }; + Some(match range_evaluation { + Some(range) if range.step_ms > 0 && range.evaluations_per_request() > 1 => { + greatest_common_divisor(times_grid_ms, range.step_ms) + } + _ => times_grid_ms, + }) + } } /// The most evaluation steps one range request may have, as Prometheus @@ -370,30 +412,11 @@ pub struct QueryWorkloadEntry { } impl QueryWorkloadEntry { - /// The grid every evaluation of this query falls on, in milliseconds, - /// anchored at the epoch (every evaluation time is a multiple of it): - /// its fixed interval ([`QueryRecurrence::fixed_interval_ms`]), or with - /// a range evaluation of more than one step the greatest common divisor - /// of the interval and the step, since each run evaluates at every step - /// up to its end; with a declared evaluation phase - /// ([`RepeatedDemand::FixedIntervalAt`]) also the greatest common - /// divisor with that phase (every 20 s at phase 5 s is on a 5 s grid). - /// `None` when it does not repeat at a positive fixed interval. + /// The grid every evaluation of this query falls on, in milliseconds + /// ([`QueryRecurrence::evaluation_grid_ms`] of its recurrence and range + /// evaluation). pub fn evaluation_grid_ms(&self) -> Option { - let interval_ms = self.recurrence.fixed_interval_ms()?; - let step_ms = match self.range_evaluation { - Some(range) if range.step_ms > 0 && range.evaluations_per_request() > 1 => { - greatest_common_divisor(interval_ms, range.step_ms) - } - _ => interval_ms, - }; - let phase_ms = match &self.recurrence { - QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { - evaluation_phase, .. - }) => evaluation_phase.0, - _ => 0, - }; - Some(greatest_common_divisor(step_ms, phase_ms)) + self.recurrence.evaluation_grid_ms(self.range_evaluation) } } From 37452e4c299a9a0511f254f3ac96c6f45170e77f Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Fri, 6 Nov 2026 19:12:14 +0000 Subject: [PATCH 2/3] test(types): delivers_on fixture built through the Memo methods (#1207 PR 1) --- .../src/stages/physical/memo_candidates.rs | 96 ++++++++++--------- 1 file changed, 51 insertions(+), 45 deletions(-) diff --git a/crates/types/src/stages/physical/memo_candidates.rs b/crates/types/src/stages/physical/memo_candidates.rs index c1807805e..3e480df36 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -200,19 +200,17 @@ pub fn output_references(query_roots: &[QueryRoot]) -> Vec<(usize, #[cfg(test)] mod tests { - use std::rc::Rc; - use super::*; - use crate::ir::operator::operator_properties::Reduction; - use crate::ir::scalar::expr_ir::{ColumnRef, ScalarValue}; + use crate::ir::operator::operator_properties::{Reduction, Source}; + use crate::ir::scalar::expr_ir::ColumnRef; use crate::ir::schema::{ DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchParams, SummaryUpdate, }; - use crate::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, ScalarExpr}; + use crate::ir::{ASAPOp, NonASAPOp, Operator}; use crate::physical::StorageMedium; use crate::stages::{ - CandidateExpression, CandidateForMemoGroup, LogicalProperties, MemoGroup, + CandidateExpression, CandidateForMemoGroup, LogicalProperties, LogicalRequirements, OriginOfCandidateForMemoGroup, }; use crate::workload::{DemandEstimate, RangeEvaluation, Rate, RepetitionInterval, TimestampMs}; @@ -220,26 +218,44 @@ mod tests { /// Candidate 0 of the test memo has no window layout. const UNWINDOWED: CandidateForMemoGroupId = CandidateForMemoGroupId { - group: MemoGroupId(0), + group: MemoGroupId(1), index_in_group: 0, }; /// Candidate 1 has tumbling windows 5 s wide. const WINDOWS_5S: CandidateForMemoGroupId = CandidateForMemoGroupId { - group: MemoGroupId(0), + group: MemoGroupId(1), index_in_group: 1, }; /// Candidate 2 has tumbling windows 10 s wide. const WINDOWS_10S: CandidateForMemoGroupId = CandidateForMemoGroupId { - group: MemoGroupId(0), + group: MemoGroupId(1), index_in_group: 2, }; - /// One group of KLL states with three candidates: no window layout, - /// 5 s tumbling windows and 10 s tumbling windows. + /// Group 1, KLL states over the scan group 0, with three candidates: + /// no window layout, 5 s tumbling windows and 10 s tumbling windows. + /// Built through `find_or_add_group` and `add_candidate` (#1207 PR 2). fn memo() -> Memo { + let mut memo = Memo::default(); + let scan = memo + .find_or_add_group(Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { + metric: "latency".into(), + }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("value", DataType::Float64, false), + ], + 0, + vec![], + ), + })) + .unwrap(); let expression = CandidateExpression { operator: Operator::ASAP(ASAPOp::SummaryAgg { - child: MemoGroupId(0), + child: scan, family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), input: SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::PerEntity, @@ -247,40 +263,30 @@ mod tests { filter: None, }), }; - let candidate = |window_layout| CandidateForMemoGroup { - logical_properties: LogicalProperties::new( - &expression, - Some(SketchParams::Kll { - k: 269, - failure_probability: None, - }), - window_layout, - ) - .unwrap(), - expression: expression.clone(), - input_requirements: vec![], - fused_groups: vec![], - origin: OriginOfCandidateForMemoGroup::Pass1, - }; - let values = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values { - rows: vec![vec![ScalarExpr::Literal(ScalarValue::Float64(1.0))]], - schema: Schema::lifted(vec![Field::plain("value", DataType::Float64, false)], None), - })) - .unwrap(); - Memo { - groups: vec![MemoGroup { - logical_node: Rc::clone(&values), - child_groups: vec![], - parents: vec![], - queries: vec![], - candidates: vec![ - candidate(None), - candidate(Some(WindowLayout::Tumbling { width_ms: 5_000 })), - candidate(Some(WindowLayout::Tumbling { width_ms: 10_000 })), - ], - }], - query_roots: vec![], + let states = memo.find_or_add_group(expression.operator.clone()).unwrap(); + for window_layout in [ + None, + Some(WindowLayout::Tumbling { width_ms: 5_000 }), + Some(WindowLayout::Tumbling { width_ms: 10_000 }), + ] { + let candidate = CandidateForMemoGroup { + logical_properties: LogicalProperties::new( + &expression, + Some(SketchParams::Kll { + k: 269, + failure_probability: None, + }), + window_layout, + ) + .unwrap(), + expression: expression.clone(), + input_requirements: vec![LogicalRequirements::new(None)], + fused_groups: vec![], + origin: OriginOfCandidateForMemoGroup::Pass1, + }; + memo.add_candidate(states, candidate).unwrap(); } + memo } fn candidate( From 12a4b96f73db0c17dedb2bdb4c1d9ed132c0e9b2 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:42:09 +0000 Subject: [PATCH 3/3] refactor(types): 2026-11-07 properties: chains as time-indexed state, grid without phase, one parameter table (#1207 PR 1) - LogicalProperties::new(expression, parameters): no window layout. - evaluation_grid_ms is None for a fixed interval without a phase; delivers_on reads w from the chain operator and uses only the grid. - A chain's output is a state with a time index (window end, keys, state). TimeRange and TimeShift accept a time-indexed state; a SummaryMerge over one TimeRange over a chain has the schema of the chain's build over that range, and finalizes and rolls up as it (the two PR 0 todos). merged_chain is replaced by summary_build and selected_from, which see through the window selection. - One per-family parameter table: combine, with pub strongest and weakest; dominates(p, f) := strongest(p, f) == Some(p). - One pub family_of_operator; sketch_algorithm_of shared with FamilyCapability::of. --- crates/accuracy/src/estimators/mod.rs | 3 +- crates/accuracy/src/propagation.rs | 78 ++-- crates/types/src/ir/operator/asap.rs | 158 +++++-- crates/types/src/ir/operator/non_asap.rs | 16 + .../src/ir/operator/tumbling_window_chain.rs | 36 +- .../types/src/ir/schema/family_capability.rs | 45 +- crates/types/src/stages/candidates.rs | 389 ++++++++++-------- crates/types/src/stages/mod.rs | 7 +- .../src/stages/physical/memo_candidates.rs | 116 +++--- crates/types/src/workload/mod.rs | 60 +-- crates/types/tests/tumbling_window_chain.rs | 195 ++++++++- 11 files changed, 737 insertions(+), 366 deletions(-) diff --git a/crates/accuracy/src/estimators/mod.rs b/crates/accuracy/src/estimators/mod.rs index 75684c1e3..05178a472 100644 --- a/crates/accuracy/src/estimators/mod.rs +++ b/crates/accuracy/src/estimators/mod.rs @@ -404,8 +404,7 @@ mod tests { FamilyCapability::of(&FieldDataType::Sketch(family, GroupingStrategy::default())); let list = parameter_list(family); for producer in list { - let properties = - LogicalProperties::new(&build, Some(producer.clone()), None).unwrap(); + let properties = LogicalProperties::new(&build, Some(producer.clone())).unwrap(); for floor in list { let requirement = LogicalRequirements::new(Some(floor.clone())); if !producer_satisfies_logical_requirements(&properties, &requirement) { diff --git a/crates/accuracy/src/propagation.rs b/crates/accuracy/src/propagation.rs index 11b5d8c02..6c33bf204 100644 --- a/crates/accuracy/src/propagation.rs +++ b/crates/accuracy/src/propagation.rs @@ -7,7 +7,6 @@ use std::collections::{HashMap, HashSet}; use std::rc::Rc; use asap_types::ir::operator::operator_properties::{BinaryOpKind, JoinKind, Reduction, Source}; -use asap_types::ir::operator::tumbling_window_chain::merged_chain; use asap_types::ir::operator::AggIntent; use asap_types::ir::properties::guarantee::{ AccuracyGuarantee, AccuracyMetric, BoundExpr, CompositionOperator, GuaranteeSource, @@ -524,10 +523,18 @@ fn rolled_up_guarantee(node: &OperatorNode) -> Option> Some(composed_guarantee(&consumer)) } +/// The chain whose windows `state` merges (its +/// [`OperatorNode::summary_build`] is a chain); `None` for any other state. +fn merged_windows_chain(state: &Rc) -> Option { + state + .summary_build() + .filter(|build| matches!(build, ASAPOp::SummaryTumblingWindowChain { .. })) +} + /// The guarantee of an estimate or finalized value read from merged -/// tumbling windows ([`merged_chain`]): the guarantee of the same read of one -/// build over the merged tumbling windows' time range (the whole chain's, or -/// its newest tumbling windows') when the +/// tumbling windows (a state whose [`OperatorNode::summary_build`] is a +/// chain): the guarantee of the same read of one build over the time range +/// the merged windows cover when the /// family's merged error for that read is exact or the same as that build's /// (the family-capability table); none when it is approximate (evidence /// only, for plan selection to judge) or the family has no merge. `None` for @@ -538,13 +545,13 @@ fn merged_tumbling_windows_guarantee(node: &OperatorNode) -> Option { - let chain = merged_chain(summary_input)?; - let family = chain.asap()?.produced_state()?; + let chain = merged_windows_chain(summary_input)?; + let family = chain.produced_state()?; FamilyCapability::of(family).merged_error(query) } ASAPOp::FinalizeExactAccumulator { child } => { - let chain = merged_chain(child)?; - let family = chain.asap()?.produced_state()?; + let chain = merged_windows_chain(child)?; + let family = chain.produced_state()?; // An exact accumulator's merged state is exact. let merged = FamilyCapability::of(family).merge_inputs; merged.map(|_| MergedError::Exact) @@ -661,40 +668,39 @@ pub(crate) fn estimates_raw_data_samples_of(node: &OperatorNode) -> Option match merged_chain(summary_input) - .unwrap_or_else(|| Rc::clone(summary_input)) - .asap()? - { - ASAPOp::SummaryAgg { - input: - SummaryUpdate { - item: None, - weight: SummaryInputExpr::Column(ColumnRef::SampleValue), - .. - }, - filter: None, - child, - .. + }) if (0.0..=1.0).contains(q) => { + match &merged_windows_chain(summary_input).or_else(|| summary_input.asap().cloned())? { + ASAPOp::SummaryAgg { + input: + SummaryUpdate { + item: None, + weight: SummaryInputExpr::Column(ColumnRef::SampleValue), + .. + }, + filter: None, + child, + .. + } + | ASAPOp::SummaryTumblingWindowChain { + input: + SummaryUpdate { + item: None, + weight: SummaryInputExpr::Column(ColumnRef::SampleValue), + .. + }, + filter: None, + child, + .. + } => estimates_raw_data_samples_of(child), + _ => None, } - | ASAPOp::SummaryTumblingWindowChain { - input: - SummaryUpdate { - item: None, - weight: SummaryInputExpr::Column(ColumnRef::SampleValue), - .. - }, - filter: None, - child, - .. - } => estimates_raw_data_samples_of(child), - _ => None, - }, + } Operator::ASAP(ASAPOp::SummaryEstimate { summary_input, .. }) => { match summary_input.asap()? { ASAPOp::SummaryAgg { diff --git a/crates/types/src/ir/operator/asap.rs b/crates/types/src/ir/operator/asap.rs index 8a2941a24..ec6cd4274 100644 --- a/crates/types/src/ir/operator/asap.rs +++ b/crates/types/src/ir/operator/asap.rs @@ -7,6 +7,8 @@ use serde::{Deserialize, Serialize}; use super::input::{InputOrTemporaryBuild, OperatorInput}; use super::node::{Operator, OperatorNode, OperatorResultKind}; +use super::non_asap::NonASAPOp; +use super::tumbling_window_chain::is_time_indexed_state; use crate::ir::operator::maintained_population::{MaintainedPopulation, PopulationStatistic}; use crate::ir::operator::operator_properties::Reduction; use crate::ir::schema::sketch::{GroupingStrategy, SketchStatistic, SummaryUpdate}; @@ -277,13 +279,13 @@ impl ASAPOp { family, reduction, .. - } - | SummaryTumblingWindowChain { + } => summary_agg_schema(child.schema(), family, reduction)?, + SummaryTumblingWindowChain { child, family, reduction, .. - } => summary_agg_schema(child.schema(), family, reduction)?, + } => chain_schema(child.schema(), family, reduction)?, SummaryEstimate { summary_input, query, @@ -332,9 +334,8 @@ impl ASAPOp { if let Some((build, schema)) = rolled_up_build(child) { return finalized_schema(Some(&build), &schema); } - if super::tumbling_window_chain::merged_chain(child).is_some() { - todo!("#1207 PR 1: finalize the windows a TimeRange over a chain selects as one build over that TimeRange") - } + // The windows a `TimeRange` over a chain selects, merged, + // finalize as the chain's build over that range. finalized_schema(summary_build(child).as_ref(), child.schema())? } MaintainPopulation { child, .. } => child.schema().clone(), @@ -409,9 +410,22 @@ impl ASAPOp { } => { self.validate_inputs()?; let input = children[0].schema(); - match group_by { - None => input.clone(), - Some(keys) => regrouped_schema(input, keys)?, + match (group_by, summary_build(&children[0])) { + (Some(keys), _) => regrouped_schema(input, keys)?, + // The windows a `TimeRange` over a chain selects merge + // into the state of the chain's build over that range, + // which has no window column (a `TimeRange` and a + // `TimeShift` keep their input's schema). + ( + None, + Some(SummaryTumblingWindowChain { + child, + family, + reduction, + .. + }), + ) => summary_agg_schema(child.schema(), &family, &reduction)?, + (None, _) => input.clone(), } } SummarySubtract { .. } @@ -473,6 +487,19 @@ impl ASAPOp { "summary merge requires exactly one state column".into(), )); } + // Over a time-indexed state: the windows one `TimeRange` + // over a chain selects, not the whole chain. + if children.iter().any(is_time_indexed_state) + && !matches!( + children.as_slice(), + [child] if matches!(child.operator(), Operator::NonASAP(NonASAPOp::TimeRange { .. })) + ) + { + return Err(SchemaDerivationError::InvalidScalarSignature( + "a summary merge over a time-indexed state reads one TimeRange over a chain" + .into(), + )); + } for child in children { needs_state(child, "SummaryMerge")?; if child.schema() != first.schema() { @@ -654,23 +681,33 @@ fn rolled_up_build( return None; }; let input = selected_from(input).unwrap_or_else(|| input.clone()); - let build = match input.operator() { - Operator::ASAP(op @ ASAPOp::SummaryAgg { .. }) => { + // A build, or the windows a `TimeRange` over a chain selects, merged: + // the chain's build over that range, whose schema is the build's over + // the chain's input. + let build = match (input.operator(), summary_build(&input)) { + (Operator::ASAP(op @ ASAPOp::SummaryAgg { .. }), _) + | (_, Some(op @ ASAPOp::SummaryTumblingWindowChain { .. })) => { op.map_children(|child| InputOrTemporaryBuild::Input(child.clone())) } - _ if super::tumbling_window_chain::merged_chain(&input).is_some() => { - todo!("#1207 PR 1: roll up the windows a TimeRange over a chain selects as one build over that TimeRange") - } _ => return None, }; - let ASAPOp::SummaryAgg { + let (ASAPOp::SummaryAgg { child, family, input: update, reduction, grouping, filter, - } = build + } + | ASAPOp::SummaryTumblingWindowChain { + child, + family, + input: update, + reduction, + grouping, + filter, + .. + }) = build else { return None; }; @@ -723,6 +760,36 @@ pub(crate) fn summary_agg_schema( Ok(schema) } +/// The schema of a `SummaryTumblingWindowChain` of `family` by `reduction` +/// over rows of schema `child`: a state with a time index, one row per +/// window, (window end, keys, state). The window end is `child`'s time +/// column; a per-entity build already keeps it. +fn chain_schema( + child: &Schema, + family: &FieldDataType, + reduction: &Reduction, +) -> Result { + let mut schema = summary_agg_schema(child, family, reduction)?; + if schema.time_index.is_none() { + let window_end = child + .time_index + .and_then(|index| child.fields.get(index)) + .ok_or_else(|| { + SchemaDerivationError::InvalidScalarSignature( + "a tumbling-window chain reads rows with a time column".into(), + ) + })?; + schema.fields.insert(0, window_end.clone()); + schema.time_index = Some(0); + // One row per window and key. + for key in &mut schema.unique_keys { + key.iter_mut().for_each(|column| *column += 1); + key.insert(0, 0); + } + } + Ok(schema) +} + /// The schema `FinalizeExactAccumulator` derives over state of /// `state_schema` that `build` (the state's `SummaryAgg`, if any) builds: /// each exact aggregate replaced by the value it finalizes to. @@ -730,14 +797,24 @@ fn finalized_schema( build: Option<&ASAPOp>, state_schema: &Schema, ) -> Result { - let value_result = if let Some(ASAPOp::SummaryAgg { - child: source, - family: FieldDataType::ExactAggregate(kind, _), - input, - reduction, - filter, - .. - }) = build + let value_result = if let Some( + ASAPOp::SummaryAgg { + child: source, + family: FieldDataType::ExactAggregate(kind, _), + input, + reduction, + filter, + .. + } + | ASAPOp::SummaryTumblingWindowChain { + child: source, + family: FieldDataType::ExactAggregate(kind, _), + input, + reduction, + filter, + .. + }, + ) = build { { use crate::ir::operator::AggIntent; @@ -845,22 +922,24 @@ fn finalized_data_type(kind: &crate::ir::schema::sketch::ExactKind) -> DataType } } -/// The states `node` selects from when it is a `Filter` that keeps whole -/// summary states by their plain columns (its output is state); `None` for +/// The states `node` selects from when it keeps whole summary states (its +/// output is state): a `Filter` by their plain columns, or a `TimeRange` or +/// `TimeShift` over a time-indexed state, which selects windows. `None` for /// any other input. pub fn selected_from(node: &I) -> Option { match node.operator() { - Operator::NonASAP(crate::ir::NonASAPOp::Filter { child, .. }) - if node.result_kind() == OperatorResultKind::State => - { - Some(child) - } + Operator::NonASAP( + NonASAPOp::Filter { child, .. } + | NonASAPOp::TimeRange { child, .. } + | NonASAPOp::TimeShift { child, .. }, + ) if node.result_kind() == OperatorResultKind::State => Some(child), _ => None, } } /// The build whose state this summary state carries: a `SummaryAgg` or -/// `SummaryTumblingWindowChain` itself, or the first input's build under +/// `SummaryTumblingWindowChain` itself, the chain under the `TimeRange` and +/// `TimeShift`s that select its windows, or the first input's build under /// `SummaryMerge`s (merged inputs have one definition, which summary /// coverage checks). `None` under a regrouping merge, whose states no build /// carries (its coverage's definition is the build it stands for), and for @@ -875,6 +954,21 @@ pub fn summary_build(state: &I) -> Option> { group_by: None, .. }) => summary_build(children.first()?), + Operator::NonASAP(NonASAPOp::TimeRange { .. } | NonASAPOp::TimeShift { .. }) => { + summary_build(&selected_from(state)?) + } + _ => None, + } +} + +/// The summary family a build operator builds: a `SummaryAgg`'s or a +/// `SummaryTumblingWindowChain`'s `family`; `None` for every other +/// operator. +pub fn family_of_operator(operator: &Operator) -> Option<&FieldDataType> { + match operator { + Operator::ASAP( + ASAPOp::SummaryAgg { family, .. } | ASAPOp::SummaryTumblingWindowChain { family, .. }, + ) => Some(family), _ => None, } } diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs index dd1755456..873443f15 100644 --- a/crates/types/src/ir/operator/non_asap.rs +++ b/crates/types/src/ir/operator/non_asap.rs @@ -8,6 +8,7 @@ use serde::{Deserialize, Serialize}; use super::input::OperatorInput; use super::node::{Operator, OperatorNode, OperatorResultKind}; +use super::tumbling_window_chain::is_time_indexed_state; use crate::ir::operator::agg_intent::AggIntent; use crate::ir::operator::operator_properties::{ BinaryOpKind, GroupKeys, InfoMatcher, JoinKind, Reduction, RelationalSetOpKind, SampleKind, @@ -921,6 +922,12 @@ impl NonASAPOp { TimeRange { child, .. } if child.result_kind() == OperatorResultKind::Relation => { OperatorResultKind::Relation } + // Over a time-indexed state (a chain): the windows it selects. + TimeRange { child, .. } | TimeShift { child, .. } + if child.result_kind() == OperatorResultKind::State => + { + OperatorResultKind::State + } TimeRange { kind, .. } => match kind { TimeRangeKind::Instant => OperatorResultKind::InstantVector, TimeRangeKind::Range => OperatorResultKind::RangeVector, @@ -1162,6 +1169,15 @@ impl NonASAPOp { return Err(invalid("label expression requires a string")); } } + TimeRange { child, .. } | TimeShift { child, .. } + if child.result_kind() == OperatorResultKind::State => + { + if !is_time_indexed_state(child) { + return Err(invalid( + "a time selection over state requires a time-indexed state", + )); + } + } TimeRange { child, .. } => { if child.result_kind() != OperatorResultKind::Relation { instant(child)?; diff --git a/crates/types/src/ir/operator/tumbling_window_chain.rs b/crates/types/src/ir/operator/tumbling_window_chain.rs index aca6285b0..065a3a310 100644 --- a/crates/types/src/ir/operator/tumbling_window_chain.rs +++ b/crates/types/src/ir/operator/tumbling_window_chain.rs @@ -1,33 +1,23 @@ //! Tumbling-window chains ([`ASAPOp::SummaryTumblingWindowChain`]): one //! row per window of width `w`, on a grid anchored at the epoch, a state //! with a time index. A chain has no range and no window count: a -//! `TimeRange` (and `TimeShift`) over it selects the windows a query reads -//! (#1207 PR 1), and how many windows it retains is $h_v$ (#1207 PR 7). +//! `TimeRange` (and `TimeShift`s) over it select the windows a query reads, +//! a `SummaryMerge` over that selection is the state of the chain's build +//! over the same range, and how many windows it retains is $h_v$ (#1207 +//! PR 7). use super::asap::ASAPOp; use super::input::OperatorInput; use super::node::Operator; +use super::non_asap::NonASAPOp; -/// The `SummaryTumblingWindowChain` a summary state carries merged: the -/// chain itself, or the one chain a `SummaryMerge` that keeps its groups -/// reads. `None` for any other input, a rollup (regrouping merge) included. -pub fn merged_chain(state: &I) -> Option { +/// Whether `state` is a state with a time index: a chain, or a `TimeRange` +/// or `TimeShift` over one. +pub(crate) fn is_time_indexed_state(state: &I) -> bool { match state.operator() { - Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. }) => Some(state.clone()), - Operator::ASAP(ASAPOp::SummaryMerge { - children, - group_by: None, - .. - }) => match children.as_slice() { - [child] - if matches!( - child.operator(), - Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. }) - ) => - { - Some(child.clone()) - } - _ => None, - }, - _ => None, + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { .. }) => true, + Operator::NonASAP( + NonASAPOp::TimeRange { child, .. } | NonASAPOp::TimeShift { child, .. }, + ) => is_time_indexed_state(&child), + _ => false, } } diff --git a/crates/types/src/ir/schema/family_capability.rs b/crates/types/src/ir/schema/family_capability.rs index 960403e36..3bcf2a57e 100644 --- a/crates/types/src/ir/schema/family_capability.rs +++ b/crates/types/src/ir/schema/family_capability.rs @@ -163,20 +163,15 @@ impl FamilyCapability { E::IRate => (None, &[]), }, // A Hydra answers per subpopulation from a shared grid of the - // inner family's cells. HydraKll merges as sketchlib's row says: - // `Hydra::merge` requires the same k and m and otherwise returns - // an error (asap_sketchlib #251). Nothing builds one yet: Pass 1 - // offers no Hydra, and the executor builds only HydraCms - // (`create_hydra_cms`). - FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind, .. }) => { - return Self::of_per_instance_sketch(&match kind { - HydraKind::HydraCms => SketchAlgorithm::HydraCms, - HydraKind::HydraCountSketch => SketchAlgorithm::HydraCountSketch, - HydraKind::HydraKll => SketchAlgorithm::HydraKll, - }) - } - FieldDataType::Sketch(algorithm, GroupingStrategy::PerSubpopulationInstance) => { - return Self::of_per_instance_sketch(algorithm) + // inner family's cells, so its row is its Hydra family's. HydraKll + // merges as sketchlib's row says: `Hydra::merge` requires the same + // k and m and otherwise returns an error (asap_sketchlib #251). + // Nothing builds one yet: Pass 1 offers no Hydra, and the executor + // builds only HydraCms (`create_hydra_cms`). + FieldDataType::Sketch(..) => { + return Self::of_per_instance_sketch( + &sketch_algorithm_of(family).expect("a sketch family has an algorithm"), + ) } FieldDataType::Plain(_) | FieldDataType::SummarySample(..) @@ -190,6 +185,28 @@ impl FamilyCapability { } } +/// The sketch algorithm a summary family builds: a per-instance sketch's +/// own, a Hydra's Hydra algorithm; `None` for a family that is not a sketch. +pub(crate) fn sketch_algorithm_of(family: &FieldDataType) -> Option { + match family { + FieldDataType::Sketch(algorithm, GroupingStrategy::PerSubpopulationInstance) => { + Some(*algorithm) + } + FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind }) => { + Some(match kind { + HydraKind::HydraCms => SketchAlgorithm::HydraCms, + HydraKind::HydraCountSketch => SketchAlgorithm::HydraCountSketch, + HydraKind::HydraKll => SketchAlgorithm::HydraKll, + }) + } + FieldDataType::Plain(_) + | FieldDataType::ExactAggregate(..) + | FieldDataType::SummarySample(..) + | FieldDataType::Wavelet(..) + | FieldDataType::StatModel(..) => None, + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/types/src/stages/candidates.rs b/crates/types/src/stages/candidates.rs index 0d5ac6100..4779ffb7f 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -6,12 +6,11 @@ use std::collections::HashSet; use super::{GroupProperties, MemoGroup, MemoGroupAsInput, MemoGroupId}; +use crate::ir::operator::asap::family_of_operator; use crate::ir::properties::summary_coverage::SummaryCoverage; use crate::ir::properties::AccuracyMetric; -use crate::ir::schema::{ - FamilyCapability, FieldDataType, GroupingStrategy, HydraKind, HydraParams, SketchAlgorithm, - SketchParams, -}; +use crate::ir::schema::family_capability::sketch_algorithm_of; +use crate::ir::schema::{HydraParams, SketchAlgorithm, SketchParams}; use crate::ir::ASAPOp; use crate::ir::Operator; use crate::ir::QueryRoot; @@ -247,7 +246,8 @@ impl LogicalProperties { expression: &CandidateExpression, parameters: Option, ) -> Result { - match (family_of_operator(&expression.operator), ¶meters) { + let family = family_of_operator(&expression.operator).and_then(sketch_algorithm_of); + match (family, ¶meters) { (Some(_), None) => return Err(LogicalPropertiesError::SketchBuildWithoutParameters), (None, Some(_)) => return Err(LogicalPropertiesError::ParametersOnExactOperator), (Some(family), Some(parameters)) => { @@ -527,35 +527,6 @@ const fn univmon_distinct(heap_size: u32) -> SketchParams { } } -/// The sketch family an operator builds: a `SummaryAgg`'s or a -/// `SummaryTumblingWindowChain`'s sketch family (a Hydra one's Hydra -/// family); `None` for every other operator, an exact accumulator build -/// included. -fn family_of_operator(operator: &Operator) -> Option { - let (Operator::ASAP(ASAPOp::SummaryAgg { family, .. }) - | Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { family, .. })) = operator - else { - return None; - }; - match family { - FieldDataType::Sketch(algorithm, GroupingStrategy::PerSubpopulationInstance) => { - Some(*algorithm) - } - FieldDataType::Sketch(_, GroupingStrategy::SharedMultiSubpopulation { kind }) => { - Some(match kind { - HydraKind::HydraKll => SketchAlgorithm::HydraKll, - HydraKind::HydraCms => SketchAlgorithm::HydraCms, - HydraKind::HydraCountSketch => SketchAlgorithm::HydraCountSketch, - }) - } - FieldDataType::Plain(_) - | FieldDataType::ExactAggregate(..) - | FieldDataType::SummarySample(..) - | FieldDataType::Wavelet(..) - | FieldDataType::StatModel(..) => None, - } -} - /// The family `parameters` size. fn family_of_parameters(parameters: &SketchParams) -> SketchAlgorithm { match parameters { @@ -576,17 +547,6 @@ fn family_of_parameters(parameters: &SketchParams) -> SketchAlgorithm { } } -/// Whether a tumbling window layout fits the operator: a `SummaryAgg` whose -/// states merge, so that its windows' states make the whole range's. -fn can_be_windowed(operator: &Operator) -> bool { - match operator { - Operator::ASAP(ASAPOp::SummaryAgg { family, .. }) => { - FamilyCapability::of(family).merge_inputs.is_some() - } - Operator::ASAP(_) | Operator::NonASAP(_) => false, - } -} - /// $r^{\log}_{c,i}$: only what $G_{c,i}$'s identity does not fix. #[derive(Debug, Clone)] pub struct LogicalRequirements { @@ -653,28 +613,81 @@ pub fn producer_satisfies_logical_requirements( } } -/// $\boldsymbol\theta \succeq \boldsymbol\theta^{\min}$: every parameter -/// that sets $\varepsilon$ is at least the floor's (at most, for DDSketch's -/// `alpha`). The recorded failure probabilities and relative errors are not -/// compared: every listed configuration of a family records the same -/// failure probability, 0.01. Parameters of different families never -/// dominate. +/// $\boldsymbol\theta \succeq \boldsymbol\theta^{\min}$: `parameters` is +/// the [`strongest`] of itself and `floor`, so every parameter that sets +/// $\varepsilon$ or holds items is at least the floor's (at most, for +/// DDSketch's `alpha`). Parameters of different families never dominate. fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { + strongest(parameters, floor).as_ref() == Some(parameters) +} + +/// The per-parameter maximum of two configurations of one family (the +/// smaller DDSketch `alpha`): what satisfies both. `None` for two +/// families. +pub fn strongest(left: &SketchParams, right: &SketchParams) -> Option { + combine(left, right, true) +} + +/// The per-parameter minimum of two configurations of one family (the +/// larger DDSketch `alpha`): a floor both satisfy. `None` for two families. +pub fn weakest(left: &SketchParams, right: &SketchParams) -> Option { + combine(left, right, false) +} + +/// [`strongest`] (`stronger`) or [`weakest`]: the one per-family table of +/// the parameters that set $\varepsilon$ or hold items, each combined. The +/// recorded error and failure probability are `left`'s: every listed +/// configuration of a family records the same failure probability, 0.01. +fn combine(left: &SketchParams, right: &SketchParams, stronger: bool) -> Option { use SketchParams as P; - match (parameters, floor) { - (P::Kll { k, .. }, P::Kll { k: floor_k, .. }) - | (P::Kmv { k, .. }, P::Kmv { k: floor_k, .. }) => k >= floor_k, - (P::DDSketch { alpha }, P::DDSketch { alpha: floor_alpha }) => alpha <= floor_alpha, + /// The larger of `a` and `b` when `larger`, else the smaller. + fn pick(larger: bool, a: T, b: T) -> T { + if (b > a) == larger { + b + } else { + a + } + } + Some(match (left, right) { + ( + P::Kll { + k, + failure_probability, + }, + P::Kll { k: other, .. }, + ) => P::Kll { + k: pick(stronger, *k, *other), + failure_probability: *failure_probability, + }, + (P::Kmv { k, relative_error }, P::Kmv { k: other, .. }) => P::Kmv { + k: pick(stronger, *k, *other), + relative_error: *relative_error, + }, + // A smaller `alpha` is stronger. + (P::DDSketch { alpha }, P::DDSketch { alpha: other }) => P::DDSketch { + alpha: pick(!stronger, *alpha, *other), + }, ( - P::Hll { precision, .. }, P::Hll { - precision: floor_precision, - .. + precision, + relative_error, + }, + P::Hll { + precision: other, .. }, - ) => precision >= floor_precision, - (P::Cms { width, depth }, P::Cms { width: w, depth: d }) - | (P::CountSketch { width, depth }, P::CountSketch { width: w, depth: d }) => { - width >= w && depth >= d + ) => P::Hll { + precision: pick(stronger, *precision, *other), + relative_error: *relative_error, + }, + (P::Cms { width, depth }, P::Cms { width: w, depth: d }) => P::Cms { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + }, + (P::CountSketch { width, depth }, P::CountSketch { width: w, depth: d }) => { + P::CountSketch { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + } } ( P::CmsWithHeap { @@ -687,8 +700,12 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { depth: d, heap_size: h, }, - ) - | ( + ) => P::CmsWithHeap { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + heap_size: pick(stronger, *heap_size, *h), + }, + ( P::CountSketchWithHeap { width, depth, @@ -699,14 +716,18 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { depth: d, heap_size: h, }, - ) => width >= w && depth >= d && heap_size >= h, + ) => P::CountSketchWithHeap { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + heap_size: pick(stronger, *heap_size, *h), + }, ( P::UnivMon { heap_size, sketch_rows, sketch_cols, layers, - .. + distinct_failure_probability, }, P::UnivMon { heap_size: h, @@ -715,14 +736,23 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { layers: l, .. }, - ) => heap_size >= h && sketch_rows >= r && sketch_cols >= c && layers >= l, + ) => P::UnivMon { + heap_size: pick(stronger, *heap_size, *h), + sketch_rows: pick(stronger, *sketch_rows, *r), + sketch_cols: pick(stronger, *sketch_cols, *c), + layers: pick(stronger, *layers, *l), + distinct_failure_probability: *distinct_failure_probability, + }, ( P::Hydra(HydraParams::HydraKll { k, shared_buckets }), P::Hydra(HydraParams::HydraKll { - k: floor_k, + k: other, shared_buckets: b, }), - ) => k >= floor_k && shared_buckets >= b, + ) => P::Hydra(HydraParams::HydraKll { + k: pick(stronger, *k, *other), + shared_buckets: pick(stronger, *shared_buckets, *b), + }), ( P::Hydra(HydraParams::HydraCms { width, @@ -736,8 +766,13 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { shared_rows: r, shared_columns: c, }), - ) - | ( + ) => P::Hydra(HydraParams::HydraCms { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + shared_rows: pick(stronger, *shared_rows, *r), + shared_columns: pick(stronger, *shared_columns, *c), + }), + ( P::Hydra(HydraParams::HydraCountSketch { width, depth, @@ -750,9 +785,14 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { shared_rows: r, shared_columns: c, }), - ) => width >= w && depth >= d && shared_rows >= r && shared_columns >= c, - // Different families. Listed by `parameters`' variant, not `_`, so a - // new variant fails to compile here until it is compared above. + ) => P::Hydra(HydraParams::HydraCountSketch { + width: pick(stronger, *width, *w), + depth: pick(stronger, *depth, *d), + shared_rows: pick(stronger, *shared_rows, *r), + shared_columns: pick(stronger, *shared_columns, *c), + }), + // Different families. Listed by `left`'s variant, not `_`, so a new + // variant fails to compile here until it is combined above. (P::Kll { .. }, _) | (P::Kmv { .. }, _) | (P::DDSketch { .. }, _) @@ -762,8 +802,8 @@ fn dominates(parameters: &SketchParams, floor: &SketchParams) -> bool { | (P::CmsWithHeap { .. }, _) | (P::CountSketchWithHeap { .. }, _) | (P::UnivMon { .. }, _) - | (P::Hydra(_), _) => false, - } + | (P::Hydra(_), _) => return None, + }) } #[cfg(test)] @@ -771,7 +811,7 @@ mod tests { use super::*; use crate::ir::operator::operator_properties::Reduction; use crate::ir::scalar::expr_ir::ColumnRef; - use crate::ir::schema::{ExactKind, SummaryUpdate}; + use crate::ir::schema::{ExactKind, FieldDataType, GroupingStrategy, HydraKind, SummaryUpdate}; /// `SummaryAgg` over group 0 building a state of `family`. fn summary_agg(family: FieldDataType) -> CandidateExpression { @@ -803,13 +843,12 @@ mod tests { shared_columns: 1_024, }); - /// `LogicalProperties::new` over every operator kind, parameters and - /// window layout: a sketch build needs parameters of its family from its - /// list; parameters are rejected on an operator without a sketch family; - /// a window is rejected at width 0 or on an operator other than a - /// `SummaryAgg` whose states merge. Parameters are checked first. + /// `LogicalProperties::new` over every operator kind and parameters: a + /// sketch build (a chain included) needs parameters of its family from + /// its list; parameters are rejected on an operator without a sketch + /// family. #[test] - fn logical_properties_new_over_every_operator_parameters_and_window() { + fn logical_properties_new_over_every_operator_and_parameters() { use LogicalPropertiesError as Invalid; let per_instance = GroupingStrategy::PerSubpopulationInstance; let hydra = GroupingStrategy::SharedMultiSubpopulation { @@ -824,7 +863,6 @@ mod tests { grouping: per_instance.clone(), filter: None, tumbling_window_width_ms: 5_000, - tumbling_window_count: 12, }), }; let readout = CandidateExpression { @@ -833,7 +871,7 @@ mod tests { query: crate::ir::schema::SketchStatistic::Quantile { q: 0.5 }, }), }; - // KLL_270 is of KLL's type but not in its list. + // kll(270) is of KLL's type but not in its list. let parameters = [ None, Some(KLL_269), @@ -841,33 +879,32 @@ mod tests { Some(HYDRA_CMS), Some(kll(270)), ]; - let windows = [ - None, - Some(WindowLayout::Tumbling { width_ms: 5_000 }), - Some(WindowLayout::Tumbling { width_ms: 0 }), - ]; let no_parameters = Err(Invalid::ParametersOnExactOperator); let needs_parameters = Err(Invalid::SketchBuildWithoutParameters); let not_in_list = Err(Invalid::ParametersNotInList); let another_family = Err(Invalid::ParametersOfAnotherFamily); - let cannot_be_windowed = Err(Invalid::WindowOnOperatorThatCannotBeWindowed); - let zero_width = Err(Invalid::ZeroWindowWidth); - // Per operator: the result for each of `parameters`, then for each of - // `windows`; a combination fails with the parameters' error first. + let exact = [ + Ok(()), + no_parameters, + no_parameters, + no_parameters, + no_parameters, + ]; + let kll_build = [ + needs_parameters, + Ok(()), + another_family, + another_family, + not_in_list, + ]; + // Per operator: the result for each of `parameters`. let table = [ ( summary_agg(sketch(SketchAlgorithm::Kll, per_instance.clone())), - [ - needs_parameters, - Ok(()), - another_family, - another_family, - not_in_list, - ], - [Ok(()), Ok(()), zero_width], + kll_build, ), + // No Hydra θ is listed yet, so no Hydra build is valid. ( - // No Hydra θ is listed yet, so no Hydra build is valid. summary_agg(sketch(SketchAlgorithm::Cms, hydra)), [ needs_parameters, @@ -876,75 +913,87 @@ mod tests { not_in_list, another_family, ], - [Ok(()), Ok(()), zero_width], - ), - ( - summary_agg(FieldDataType::exact(ExactKind::Sum)), - [ - Ok(()), - no_parameters, - no_parameters, - no_parameters, - no_parameters, - ], - [Ok(()), Ok(()), zero_width], - ), - // IRate's states do not merge across a window boundary. - ( - summary_agg(FieldDataType::exact(ExactKind::IRate)), - [ - Ok(()), - no_parameters, - no_parameters, - no_parameters, - no_parameters, - ], - [Ok(()), cannot_be_windowed, zero_width], - ), - ( - kll_chain, - [ - needs_parameters, - Ok(()), - another_family, - another_family, - not_in_list, - ], - [Ok(()), cannot_be_windowed, zero_width], - ), - ( - readout, - [ - Ok(()), - no_parameters, - no_parameters, - no_parameters, - no_parameters, - ], - [Ok(()), cannot_be_windowed, zero_width], ), + (summary_agg(FieldDataType::exact(ExactKind::Sum)), exact), + (kll_chain, kll_build), + (readout, exact), ]; - for (expression, by_parameters, by_window) in table { - for (parameters, parameters_result) in parameters.iter().zip(by_parameters) { - for (window, window_result) in windows.iter().zip(by_window) { - let expected = parameters_result.and(window_result); - let properties = - LogicalProperties::new(&expression, parameters.clone(), *window); - assert_eq!( - properties.as_ref().map(|_| ()).map_err(|e| *e), - expected, - "{:?} {parameters:?} {window:?}", - expression.operator.kind_name() - ); - if let Ok(properties) = properties { - assert_eq!(properties.parameters(), parameters.as_ref()); - assert_eq!(properties.window_layout(), window.as_ref()); - } + for (expression, by_parameters) in table { + for (parameters, expected) in parameters.iter().zip(by_parameters) { + let properties = LogicalProperties::new(&expression, parameters.clone()); + assert_eq!( + properties.as_ref().map(|_| ()).map_err(|e| *e), + expected, + "{:?} {parameters:?}", + expression.operator.kind_name() + ); + if let Ok(properties) = properties { + assert_eq!(properties.parameters(), parameters.as_ref()); } } } } + /// `family_of_operator` reads the family of a `SummaryAgg` and of a + /// chain, and of no other operator. + #[test] + fn family_of_operator_reads_the_builds_family() { + let kll = sketch(SketchAlgorithm::Kll, GroupingStrategy::default()); + let chain = Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + child: MemoGroupId(0), + family: kll.clone(), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + tumbling_window_width_ms: 5_000, + }); + let merge = Operator::ASAP(ASAPOp::SummaryMerge { + children: vec![MemoGroupId(0)], + group_by: None, + }); + assert_eq!( + family_of_operator(&summary_agg(kll.clone()).operator), + Some(&kll) + ); + assert_eq!(family_of_operator(&chain), Some(&kll)); + assert_eq!(family_of_operator(&merge), None); + } + + /// `strongest` and `weakest` take the per-parameter maximum and minimum + /// of two configurations of one family (DDSketch's smaller `alpha` is the + /// stronger), keep the left one's recorded fields, and give `None` for + /// two families. + #[test] + fn strongest_and_weakest_combine_one_family_per_parameter() { + let cms_heap = |width, depth, heap_size| SketchParams::CmsWithHeap { + width, + depth, + heap_size, + }; + let dd = |alpha| SketchParams::DDSketch { alpha }; + let hll = |precision, relative_error| SketchParams::Hll { + precision, + relative_error: Some(relative_error), + }; + for (left, right, strong, weak) in [ + (kll(52), KLL_269, kll(269), kll(52)), + (dd(0.05), dd(0.01), dd(0.01), dd(0.05)), + (hll(13, 0.1), hll(18, 0.01), hll(18, 0.1), hll(13, 0.1)), + ( + cms_heap(272, 3, 50), + cms_heap(55, 5, 10), + cms_heap(272, 5, 50), + cms_heap(55, 3, 10), + ), + ] { + assert_eq!(strongest(&left, &right), Some(strong), "{left:?} {right:?}"); + assert_eq!(weakest(&left, &right), Some(weak), "{left:?} {right:?}"); + } + assert_eq!(strongest(&KLL_269, &CMS), None); + assert_eq!(weakest(&CMS, &HYDRA_CMS), None); + } + /// Each predefined list holds only its family's parameters, without /// repeats, and is empty exactly for the families Pass 1 does not offer. #[test] @@ -1040,17 +1089,13 @@ mod tests { LogicalProperties::new( &summary_agg(sketch(family, GroupingStrategy::default())), Some(parameters.clone()), - None, ) .unwrap() }; let floor = |parameters: &SketchParams| LogicalRequirements::new(Some(parameters.clone())); - let exact = LogicalProperties::new( - &summary_agg(FieldDataType::exact(ExactKind::Sum)), - None, - None, - ) - .unwrap(); + let exact = + LogicalProperties::new(&summary_agg(FieldDataType::exact(ExactKind::Sum)), None) + .unwrap(); let no_floor = LogicalRequirements::new(None); assert!(producer_satisfies_logical_requirements( &build(kll_26), diff --git a/crates/types/src/stages/mod.rs b/crates/types/src/stages/mod.rs index f032bacad..89f454ada 100644 --- a/crates/types/src/stages/mod.rs +++ b/crates/types/src/stages/mod.rs @@ -16,9 +16,10 @@ mod pruning; mod realizer; pub use candidates::{ - parameter_list, producer_satisfies_logical_requirements, CandidateError, CandidateExpression, - CandidateForMemoGroup, CandidateForMemoGroupId, ErrorPropagation, LogicalProperties, - LogicalPropertiesError, LogicalRequirements, Memo, OriginOfCandidateForMemoGroup, + parameter_list, producer_satisfies_logical_requirements, strongest, weakest, CandidateError, + CandidateExpression, CandidateForMemoGroup, CandidateForMemoGroupId, ErrorPropagation, + LogicalProperties, LogicalPropertiesError, LogicalRequirements, Memo, + OriginOfCandidateForMemoGroup, }; pub use logical_memo_candidates::{ LogicalASAPMemoCandidates, LogicalEdge, PruneCause, PrunedLogicalCandidate, RootPruneReason, diff --git a/crates/types/src/stages/physical/memo_candidates.rs b/crates/types/src/stages/physical/memo_candidates.rs index 3e480df36..2cd89b515 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -9,11 +9,9 @@ use std::collections::HashMap; use super::execution_context::ExecutionContext; use super::graph::PhysicalCandidateId; use crate::ir::QueryRoot; +use crate::ir::{ASAPOp, Operator}; use crate::physical::{ExecutionTiming, Lifetime, PhysicalProperties}; -use crate::stages::{ - CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, MemoGroupId, PrunedCandidate, -}; -use crate::workload::{QueryRecurrence, RepeatedDemand}; +use crate::stages::{CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, PrunedCandidate}; /// Index into [`PhysicalASAPMemoCandidates::edges`]. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] @@ -31,40 +29,35 @@ pub struct PhysicalCandidate { impl PhysicalCandidate { /// Whether a consumer running in `context` can read this producer (the /// concepts doc's alignment table, section 2). True for $\ell$ = None, - /// and for `OneEvaluation` in the same context. A chain producer - /// (`SummaryTumblingWindowChain`) with $\ell$ = `AcrossEvaluations`, with - /// $w$ the chain operator's width in its expression in `memo`, needs every evaluation time of the context on its - /// grid: $w$ divides the context's evaluation grid (interval, range step - /// and phase), each scheduled time, or a one-off query's time; false - /// when the times are not known (an interval without a phase, an - /// estimated rate, an unknown recurrence). - /// - /// A `None` or `OneEvaluation` producer runs in the consumer's own - /// context ($x_{\kappa,v}$, Stage 3), so its result belongs to that - /// context. A non-windowed `AcrossEvaluations` producer is not offered - /// yet (it needs its own delivery rule), so this returns false for it. + /// and for `OneEvaluation` in the same context: such a producer runs in + /// the consumer's own context ($x_{\kappa,v}$, Stage 3). A chain producer + /// (`SummaryTumblingWindowChain`) with $\ell$ = `AcrossEvaluations`, $w$ + /// its chain operator's width in `memo`, needs every evaluation time of + /// the context on its grid: $w$ divides the context's + /// [`evaluation_grid_ms`](crate::workload::QueryRecurrence::evaluation_grid_ms); false when that is `None` + /// (the times are not known). `AcrossEvaluations` is offered only to + /// chains, so this is false for any other `AcrossEvaluations` producer. pub fn delivers_on(&self, memo: &Memo, context: &ExecutionContext) -> bool { match self.physical_properties.lifetime() { Lifetime::None | Lifetime::OneEvaluation(_) => true, Lifetime::AcrossEvaluations(_) => { - let Some(WindowLayout::Tumbling { width_ms }) = memo + let Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { + tumbling_window_width_ms, + .. + }) = memo .candidate(self.logical_candidate_id) - .logical_properties - .window_layout() + .expression + .operator else { return false; }; match context { - ExecutionContext::Evaluation { - recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(_)), - .. - } => false, ExecutionContext::Evaluation { recurrence, range_evaluation, } => recurrence .evaluation_grid_ms(*range_evaluation) - .is_some_and(|grid_ms| grid_ms.is_multiple_of(*width_ms)), + .is_some_and(|grid_ms| grid_ms.is_multiple_of(tumbling_window_width_ms)), ExecutionContext::IngestionPipeline(_) | ExecutionContext::Materialized(_) => { todo!("#1207 PR 7: delivery into an ingestion or materialized context") } @@ -207,34 +200,38 @@ mod tests { DataType, Field, FieldDataType, GroupingStrategy, Schema, SketchAlgorithm, SketchParams, SummaryUpdate, }; - use crate::ir::{ASAPOp, NonASAPOp, Operator}; + use crate::ir::NonASAPOp; use crate::physical::StorageMedium; use crate::stages::{ CandidateExpression, CandidateForMemoGroup, LogicalProperties, LogicalRequirements, - OriginOfCandidateForMemoGroup, + MemoGroupId, OriginOfCandidateForMemoGroup, + }; + use crate::workload::{ + DemandEstimate, QueryRecurrence, RangeEvaluation, Rate, RepeatedDemand, RepetitionInterval, + TimestampMs, }; - use crate::workload::{DemandEstimate, RangeEvaluation, Rate, RepetitionInterval, TimestampMs}; use ExecutionTiming::{IngestionTime, QueryTime}; - /// Candidate 0 of the test memo has no window layout. + /// The KLL build over the scan: not a chain. const UNWINDOWED: CandidateForMemoGroupId = CandidateForMemoGroupId { group: MemoGroupId(1), index_in_group: 0, }; - /// Candidate 1 has tumbling windows 5 s wide. + /// The chain of tumbling windows 5 s wide over the scan. const WINDOWS_5S: CandidateForMemoGroupId = CandidateForMemoGroupId { - group: MemoGroupId(1), - index_in_group: 1, + group: MemoGroupId(2), + index_in_group: 0, }; - /// Candidate 2 has tumbling windows 10 s wide. + /// The chain of tumbling windows 10 s wide over the scan. const WINDOWS_10S: CandidateForMemoGroupId = CandidateForMemoGroupId { - group: MemoGroupId(1), - index_in_group: 2, + group: MemoGroupId(3), + index_in_group: 0, }; - /// Group 1, KLL states over the scan group 0, with three candidates: - /// no window layout, 5 s tumbling windows and 10 s tumbling windows. - /// Built through `find_or_add_group` and `add_candidate` (#1207 PR 2). + /// Group 1, KLL states over the scan group 0, and groups 2 and 3, chains + /// of KLL states over it 5 s and 10 s wide, each with its own expression + /// as its one candidate. Built through `find_or_add_group` and + /// `add_candidate` (#1207 PR 2). fn memo() -> Memo { let mut memo = Memo::default(); let scan = memo @@ -253,22 +250,29 @@ mod tests { ), })) .unwrap(); - let expression = CandidateExpression { - operator: Operator::ASAP(ASAPOp::SummaryAgg { + let kll = FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()); + let build = Operator::ASAP(ASAPOp::SummaryAgg { + child: scan, + family: kll.clone(), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::PerEntity, + grouping: GroupingStrategy::default(), + filter: None, + }); + let chain = |tumbling_window_width_ms| { + Operator::ASAP(ASAPOp::SummaryTumblingWindowChain { child: scan, - family: FieldDataType::Sketch(SketchAlgorithm::Kll, GroupingStrategy::default()), + family: kll.clone(), input: SummaryUpdate::column(ColumnRef::SampleValue), reduction: Reduction::PerEntity, grouping: GroupingStrategy::default(), filter: None, - }), + tumbling_window_width_ms, + }) }; - let states = memo.find_or_add_group(expression.operator.clone()).unwrap(); - for window_layout in [ - None, - Some(WindowLayout::Tumbling { width_ms: 5_000 }), - Some(WindowLayout::Tumbling { width_ms: 10_000 }), - ] { + for operator in [build, chain(5_000), chain(10_000)] { + let group = memo.find_or_add_group(operator.clone()).unwrap(); + let expression = CandidateExpression { operator }; let candidate = CandidateForMemoGroup { logical_properties: LogicalProperties::new( &expression, @@ -276,15 +280,14 @@ mod tests { k: 269, failure_probability: None, }), - window_layout, ) .unwrap(), - expression: expression.clone(), + expression, input_requirements: vec![LogicalRequirements::new(None)], fused_groups: vec![], origin: OriginOfCandidateForMemoGroup::Pass1, }; - memo.add_candidate(states, candidate).unwrap(); + memo.add_candidate(group, candidate).unwrap(); } memo } @@ -320,13 +323,14 @@ mod tests { const MEMORY: StorageMedium = StorageMedium::Memory; /// `delivers_on` follows the alignment table, one case per recurrence: a - /// windowed `AcrossEvaluations` producer of width $w$ (the window - /// layout's, grid at phase 0) needs $w$ to divide the context's + /// chain producer with `AcrossEvaluations` of width $w$ (the chain + /// operator's, grid at phase 0) needs $w$ to divide the context's /// evaluation grid (interval, range step and phase), each scheduled time, /// or a one-off query's time, and is refused when the times are not - /// known; a `None` or `OneEvaluation` producer runs in the consumer's - /// context and delivers on any; a non-windowed `AcrossEvaluations` - /// producer is not offered. + /// known (a fixed interval without a phase included); a `None` or + /// `OneEvaluation` producer runs in the consumer's context and delivers + /// on any; an `AcrossEvaluations` producer that is not a chain is not + /// offered. #[test] fn delivers_on_follows_the_alignment_table() { let memo = memo(); @@ -429,7 +433,7 @@ mod tests { /// consumer state: a query-time producer serves only query-time /// consumers, an ingestion-time producer that is not materialized only /// ingestion-time consumers, and one materialized across evaluations - /// serves both. Window layouts do not matter: alignment is Stage 3's + /// serves both. Chain widths do not matter: alignment is Stage 3's /// per-context check. #[test] fn producer_satisfies_physical_requirements_over_every_valid_timing_and_lifetime() { diff --git a/crates/types/src/workload/mod.rs b/crates/types/src/workload/mod.rs index 407ca58a3..540ffeffb 100644 --- a/crates/types/src/workload/mod.rs +++ b/crates/types/src/workload/mod.rs @@ -212,22 +212,20 @@ impl QueryRecurrence { /// The grid every evaluation with this recurrence and `range_evaluation` /// falls on, in milliseconds, anchored at the epoch (every evaluation - /// time is a multiple of it). The times' grid is the fixed interval - /// ([`Self::fixed_interval_ms`]; with a declared - /// [`RepeatedDemand::FixedIntervalAt`] phase, the greatest common divisor - /// with that phase: every 20 s at phase 5 s is on a 5 s grid), the - /// greatest common divisor of the scheduled times, or a one-off query's - /// time. A range evaluation of more than one step also takes the greatest - /// common divisor with its step, since each run evaluates at every step - /// up to its end. `None` when the times are not known: an estimated + /// time is a multiple of it). The times' grid is the greatest common + /// divisor of a [`RepeatedDemand::FixedIntervalAt`]'s interval and phase + /// (every 20 s at phase 5 s is on a 5 s grid), of the scheduled times, or + /// a one-off query's time. A range evaluation of more than one step also + /// takes the greatest common divisor with its step, since each run + /// evaluates at every step up to its end. `None` when the times are not + /// known: a fixed interval without a declared phase (Prometheus staggers + /// rule groups by a hash, so it does not mean phase 0), an estimated /// rate, a one-off query without a time, an unknown recurrence, or a - /// zero interval. A fixed interval without a declared phase is taken at - /// phase 0. + /// zero interval. Pass 2's chain widths and + /// [`PhysicalCandidate::delivers_on`](crate::stages::PhysicalCandidate::delivers_on) + /// both read it. pub fn evaluation_grid_ms(&self, range_evaluation: Option) -> Option { let times_grid_ms = match self { - QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(_)) => { - self.fixed_interval_ms()? - } QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { evaluation_phase, .. }) => greatest_common_divisor(self.fixed_interval_ms()?, evaluation_phase.0), @@ -238,7 +236,9 @@ impl QueryRecurrence { execute_at: Some(time), .. } => time.0, - QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(_)) + QueryRecurrence::Repeated( + RepeatedDemand::FixedInterval(_) | RepeatedDemand::EstimatedRate(_), + ) | QueryRecurrence::OneTime { execute_at: None, .. } @@ -1497,19 +1497,27 @@ mod tests { assert_eq!(range.evaluations_per_request(), 61); let entry = QueryWorkloadEntry::from(&hour); assert_eq!(entry.range_evaluation, Some(range)); - // Refreshes every 30 s and steps every 60 s: evaluations on a 30 s grid. - assert_eq!(entry.evaluation_grid_ms(), Some(30_000)); - assert_eq!( - QueryWorkloadEntry::from(&plain).evaluation_grid_ms(), - Some(30_000) - ); + // A fixed interval without a declared phase: the evaluation times + // are not known, so there is no grid. + assert_eq!(entry.evaluation_grid_ms(), None); + assert_eq!(QueryWorkloadEntry::from(&plain).evaluation_grid_ms(), None); + let every = |interval_ms, phase_ms| { + QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { + interval: RepetitionInterval(interval_ms), + evaluation_phase: TimestampMs(phase_ms), + }) + }; + // Refreshes every 30 s at phase 0 and steps every 60 s: evaluations + // on a 30 s grid. + let stepped = QueryWorkloadEntry { + recurrence: every(30_000, 0), + ..entry.clone() + }; + assert_eq!(stepped.evaluation_grid_ms(), Some(30_000)); // At a declared phase, also on that phase from the epoch: every // minute at phase 125 s is on a 5 s grid; at phase 120 s, on 60 s. let at_phase = |phase_ms| QueryWorkloadEntry { - recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt { - interval: RepetitionInterval(60_000), - evaluation_phase: TimestampMs(phase_ms), - }), + recurrence: every(60_000, phase_ms), range_evaluation: None, ..entry.clone() }; @@ -1523,9 +1531,7 @@ mod tests { step_ms: 45_000, span_ms: 0, }), - recurrence: QueryRecurrence::Repeated(RepeatedDemand::FixedInterval( - RepetitionInterval(60_000), - )), + recurrence: every(60_000, 0), ..entry.clone() }; assert_eq!(one_step.evaluation_grid_ms(), Some(60_000)); diff --git a/crates/types/tests/tumbling_window_chain.rs b/crates/types/tests/tumbling_window_chain.rs index 439a87edb..ffc4f1a36 100644 --- a/crates/types/tests/tumbling_window_chain.rs +++ b/crates/types/tests/tumbling_window_chain.rs @@ -7,7 +7,9 @@ use asap_types::ir::operator::operator_properties::{Reduction, Source}; use asap_types::ir::scalar::ColumnRef; use asap_types::ir::schema::{DataType, Field, FieldDataType, Schema}; use asap_types::ir::schema::{GroupingStrategy, SketchAlgorithm, SummaryUpdate}; -use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, TimeRangeKind}; +use asap_types::ir::{ + ASAPOp, NonASAPOp, Operator, OperatorNode, OperatorResultKind, TimeRangeKind, +}; const MINUTE: u64 = 60_000; @@ -87,3 +89,194 @@ fn names_before_the_tumbling_window_rename_still_read() { operator ); } + +/// A scan with a `host` label, then the chain, the plain build and the +/// time selections over it by host. +fn scan_with_host() -> Rc { + OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Field::plain("ts", DataType::Timestamp, false), + Field::plain("host", DataType::Utf8, false), + Field::plain("value", DataType::Float64, false), + ], + 0, + vec![], + ), + })) + .unwrap() +} + +fn node(operator: Operator) -> Rc { + let node = OperatorNode::new_shared(operator).unwrap(); + node.validate_structure().unwrap(); + node +} + +/// The `SummaryAgg` (`chain` false) or the 1-min chain (`chain` true) of +/// `family` by host over `input`. +fn by_host(input: Rc, family: FieldDataType, chain: bool) -> Operator { + let reduction = Reduction::by(vec![1]); + let update = SummaryUpdate::column(ColumnRef::SampleValue); + let grouping = GroupingStrategy::default(); + Operator::ASAP(if chain { + ASAPOp::SummaryTumblingWindowChain { + child: input, + family, + input: update, + reduction, + grouping, + filter: None, + tumbling_window_width_ms: MINUTE, + } + } else { + ASAPOp::SummaryAgg { + child: input, + family, + input: update, + reduction, + grouping, + filter: None, + } + }) +} + +/// `TimeRange(5m, TimeShift(1h, input))`. +fn five_minutes_an_hour_ago(input: Rc) -> Rc { + let shifted = node(Operator::NonASAP(NonASAPOp::TimeShift { + shift: asap_types::ir::operator::operator_properties::TimeShift { + offset_ms: 60 * MINUTE as i64, + at: None, + }, + child: input, + })); + node(Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_millis(5 * MINUTE), + kind: TimeRangeKind::Range, + child: shifted, + })) +} + +fn merge(children: Vec>, group_by: Option>) -> Operator { + Operator::ASAP(ASAPOp::SummaryMerge { children, group_by }) +} + +/// A chain's output is a state with a time index, one row per window: +/// (window end, keys, state) by host, and the per-series build's own time +/// column per series. The plain build by host has no time index. +#[test] +fn a_chain_is_a_state_with_a_time_index() { + let by_host_chain = node(by_host(scan_with_host(), kll(), true)); + assert_eq!(by_host_chain.result_kind, OperatorResultKind::State); + assert_eq!(by_host_chain.schema.time_index, Some(0)); + let names: Vec<_> = by_host_chain + .schema + .fields + .iter() + .map(|f| f.name.as_str()) + .collect(); + assert_eq!(names, ["ts", "host", "state"]); + let build = node(by_host(scan_with_host(), kll(), false)); + assert_eq!(build.schema.time_index, None); + let per_series = chain(scan()); + assert_eq!(per_series.result_kind, OperatorResultKind::State); + assert_eq!(per_series.schema.time_index, Some(0)); +} + +/// `TimeRange` and `TimeShift` accept a time-indexed state, a chain or a +/// selection over one, and keep it a state with the chain's schema; over +/// a state without a time index (a plain build) they are rejected. +#[test] +fn time_range_and_time_shift_select_a_chains_windows() { + let windows = node(by_host(scan_with_host(), kll(), true)); + let selected = five_minutes_an_hour_ago(windows.clone()); + assert_eq!(selected.result_kind, OperatorResultKind::State); + assert_eq!(selected.schema, windows.schema); + let build = node(by_host(scan_with_host(), kll(), false)); + let over_build = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange { + range: Duration::from_millis(5 * MINUTE), + kind: TimeRangeKind::Range, + child: build, + })) + .unwrap(); + assert!(over_build.validate_structure().is_err()); +} + +/// Merging the windows `TimeRange(5m, TimeShift(1h, chain(X)))` selects +/// gives the state of `SummaryAgg(TimeRange(5m, TimeShift(1h, X)))`: the +/// same schema and kind, and so the same schema read out, finalized +/// (an exact sum) and rolled up. +#[test] +fn merged_windows_are_the_build_over_the_selected_range() { + let sum = FieldDataType::exact(asap_types::ir::schema::ExactKind::Sum); + for family in [kll(), sum.clone()] { + let merged = node(merge( + vec![five_minutes_an_hour_ago(node(by_host( + scan_with_host(), + family.clone(), + true, + )))], + None, + )); + let build = node(by_host( + five_minutes_an_hour_ago(scan_with_host()), + family.clone(), + false, + )); + assert_eq!(merged.schema, build.schema, "{family:?}"); + assert_eq!(merged.result_kind, build.result_kind); + let read = |state: &Rc| { + node(if family == sum { + Operator::ASAP(ASAPOp::FinalizeExactAccumulator { + child: state.clone(), + }) + } else { + Operator::ASAP(ASAPOp::SummaryEstimate { + summary_input: state.clone(), + query: asap_types::ir::schema::SketchStatistic::Quantile { q: 0.95 }, + }) + }) + }; + let (read_merged, read_build) = (read(&merged), read(&build)); + assert_eq!(read_merged.schema, read_build.schema, "{family:?}"); + assert_eq!(read_merged.result_kind, read_build.result_kind); + // Rolled up to one state over every host. + let (rolled_merged, rolled_build) = ( + node(merge(vec![merged], Some(vec![]))), + node(merge(vec![build], Some(vec![]))), + ); + assert_eq!(rolled_merged.schema, rolled_build.schema); + let (read_merged, read_build) = (read(&rolled_merged), read(&rolled_build)); + assert_eq!(read_merged.schema, read_build.schema, "{family:?}"); + } +} + +/// A merge over a time-indexed state reads one `TimeRange` over a chain: +/// not the whole chain, not a `TimeShift` without a range, and not beside +/// another input. +#[test] +fn a_merge_over_a_chain_reads_one_time_range() { + let windows = || node(by_host(scan_with_host(), kll(), true)); + let shifted = node(Operator::NonASAP(NonASAPOp::TimeShift { + shift: asap_types::ir::operator::operator_properties::TimeShift { + offset_ms: MINUTE as i64, + at: None, + }, + child: windows(), + })); + for children in [ + vec![windows()], + vec![shifted], + vec![ + five_minutes_an_hour_ago(windows()), + five_minutes_an_hour_ago(windows()), + ], + ] { + assert!(OperatorNode::new_shared(merge(children, None)).is_err()); + } + assert!( + OperatorNode::new_shared(merge(vec![five_minutes_an_hour_ago(windows())], None)).is_ok() + ); +}