From 5c452beaa23ebd4d7874295616fa3f1fa96e6b42 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/2] 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 | 848 +++++++++++++++++- .../src/stages/physical/memo_candidates.rs | 322 ++++++- crates/types/src/workload/mod.rs | 69 +- 12 files changed, 1434 insertions(+), 106 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 04c78b7d5..dfad7cdee 100644 --- a/crates/types/src/stages/candidates.rs +++ b/crates/types/src/stages/candidates.rs @@ -7,7 +7,10 @@ use std::collections::HashSet; use super::{GroupProperties, MemoGroup, MemoGroupAsInput, MemoGroupId}; use crate::ir::properties::summary_coverage::SummaryCoverage; -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; @@ -75,8 +78,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] } /// The groups, each after every group it or one of its candidates @@ -227,8 +229,31 @@ impl LogicalProperties { parameters: Option, window_layout: Option, ) -> Result { - let _ = (expression, parameters, window_layout); - todo!("#1207 PR 1: reject invalid parameters and invalid windows") + 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) => {} + } + if let Some(WindowLayout::Tumbling { width_ms }) = window_layout { + if width_ms == 0 { + return Err(LogicalPropertiesError::ZeroWindowWidth); + } + if !can_be_windowed(&expression.operator) { + return Err(LogicalPropertiesError::WindowOnOperatorThatCannotBeWindowed); + } + } + Ok(Self { + parameters, + window_layout, + }) } /// $\boldsymbol\theta_v$. @@ -270,9 +295,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. @@ -327,6 +648,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 5b64ece64..f1777af16 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -9,10 +9,12 @@ use std::collections::HashMap; use super::execution_context::ExecutionContext; use super::graph::PhysicalCandidateId; use crate::ir::QueryRoot; -use crate::physical::PhysicalProperties; +use crate::physical::{ExecutionTiming, Lifetime, PhysicalProperties}; use crate::stages::{ CandidateForMemoGroupId, LogicalASAPMemoCandidates, Memo, MemoGroupId, PrunedCandidate, + WindowLayout, }; +use crate::workload::{QueryRecurrence, RepeatedDemand}; /// Index into [`PhysicalASAPMemoCandidates::edges`]. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)] @@ -37,9 +39,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") + } + } + } + } } } @@ -50,8 +82,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. @@ -158,3 +202,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 fe7dec720..e6752ee84 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 b2508523e7112c06ccf377f2ae5216b0096e8805 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/2] 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 f1777af16..3eeff0d8e 100644 --- a/crates/types/src/stages/physical/memo_candidates.rs +++ b/crates/types/src/stages/physical/memo_candidates.rs @@ -205,19 +205,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}; @@ -225,26 +223,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, @@ -252,40 +268,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(