From cbb29c6f909fccc88bfa2ff2b0ba0131c7951362 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 17:54:45 +0000 Subject: [PATCH 1/2] refactor!: name summary builds by shape; one RowWeight for keyed builds Rename the executor's three summary builds by what each holds: SummaryBuild -> PerGroupSummaryBuild, KeyedSummaryBuild -> PerGroupKeyedSummaryBuild, SharedSummaryBuild -> AllGroupsSummaryBuild, with their constructors (per_group_summary_build, per_group_keyed_summary_build, all_groups_summary_build). Generalize HydraRowWeight to RowWeight { Column, Constant } and use it in the keyed per-group build too: it now accepts a constant weight (1 for a row count), an Int64 weight column and a nullable weight column (NULL adds nothing). The query-time physical planner binds a keyed heap over the unit count, which it refused before, and the raw-sample precompute adds the constant without projecting a column of ones. The i32::MAX bound stays in the all-groups (HydraCms) constructor; hydra_constant_weight becomes constant_row_weight. BREAKING CHANGE: serialized executor Operator variants are renamed, and the keyed build's `value` field is replaced by `row_weight`. --- crates/executor/src/capability.rs | 19 +- crates/executor/src/operators/mod.rs | 38 +- crates/executor/src/operators/summary/mod.rs | 138 +++++--- crates/executor/src/operators/unchecked.rs | 20 +- crates/executor/src/physical_planner/mod.rs | 54 +-- .../src/physical_planner/precompute.rs | 25 +- crates/executor/tests/blocking_resources.rs | 9 +- crates/executor/tests/current_series_heap.rs | 9 +- crates/executor/tests/hydra_cms_execution.rs | 10 +- .../tests/keyed_summary_row_weight.rs | 326 ++++++++++++++++++ crates/executor/tests/physical_dag.rs | 19 +- crates/executor/tests/physical_semantics.rs | 17 +- crates/executor/tests/univmon_execution.rs | 4 +- .../executor/tests/weighted_topk_binding.rs | 6 +- .../tests/kll_pane_execution.rs | 3 +- .../tests/precompute_raw_samples.rs | 56 +++ 16 files changed, 619 insertions(+), 134 deletions(-) create mode 100644 crates/executor/tests/keyed_summary_row_weight.rs diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index d57afb8e..d6127eb0 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -148,7 +148,9 @@ fn validate_hydra_cms( match (&input.weight, proof) { (SummaryInputExpr::Constant(w), NonNegativeWeightProof::UnitCount) if *w == 1.0 => Ok(()), (SummaryInputExpr::Constant(w), NonNegativeWeightProof::PositiveConstant) => { - hydra_constant_weight(*w).map(|_| ()) + constant_row_weight(*w) + .map(|_| ()) + .map_err(|error| format!("HydraCms {error}")) } (SummaryInputExpr::Column(_), _) => Ok(()), _ => Err( @@ -159,16 +161,17 @@ fn validate_hydra_cms( } } -/// A HydraCms constant weight as a positive integer at most `i32::MAX` -/// (asap_sketchlib's Hydra adds `i32` weights); an error otherwise. -pub(crate) fn hydra_constant_weight(weight: f64) -> Result { - if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(i32::MAX) +/// A `SummaryUpdate` constant weight as the positive integer a +/// [`RowWeight::Constant`](crate::operators::RowWeight::Constant) holds +/// (at most `u32::MAX`); an error otherwise. The all-groups (HydraCms) build +/// further bounds it by `i32::MAX`. +pub(crate) fn constant_row_weight(weight: f64) -> Result { + if weight.is_finite() && weight > 0.0 && weight.fract() == 0.0 && weight <= f64::from(u32::MAX) { Ok(std::num::NonZeroU32::new(weight as u32).expect("weight > 0")) } else { Err(format!( - "HydraCms constant weight must be a positive integer within i32 (asap_sketchlib's \ - Hydra adds i32 weights), got {weight}" + "constant weight must be a positive integer within u32, got {weight}" )) } } @@ -285,7 +288,7 @@ pub fn validate_native_family(family: &SummaryFamilyType) -> Result<(), Error> { if let SummaryFamilyType::Sketch(kind, grouping) = family { // Plain Count-Min is native as stored state only: it merges and reads // its bare count, but the DAG does not build it from rows. HydraCms - // is built by the shared summary operator. + // is built by the all-groups summary build. if let (A::Cms, SketchParams::Cms { width, depth }) = (kind.algorithm(), kind.params()) { if grouping != &GroupingStrategy::PerSubpopulationInstance { return hydra_cms_shape(family).map(|_| ()).map_err(Error::Invalid); diff --git a/crates/executor/src/operators/mod.rs b/crates/executor/src/operators/mod.rs index 1bc2a0fc..6e06c820 100644 --- a/crates/executor/src/operators/mod.rs +++ b/crates/executor/src/operators/mod.rs @@ -34,7 +34,7 @@ pub(crate) mod vector_window; pub use aggregate::Reduction; pub use series_window::SubquerySteps; pub use sort::SortKey; -pub use summary::{HydraRowWeight, SummaryEvaluation}; +pub use summary::{RowWeight, SummaryEvaluation}; #[derive(Clone, serde::Serialize, serde::Deserialize)] enum Kind { #[serde(skip)] @@ -135,7 +135,9 @@ enum Kind { kind: planner_types::ir::operator::JoinKind, predicate: Box, }, - SummaryBuild { + /// One summary state per group, updated from one value column per row + /// (built by [`Operator::per_group_summary_build`]). + PerGroupSummaryBuild { family: SummaryFamilyType, value: usize, time: Option, @@ -144,21 +146,25 @@ enum Kind { #[serde(default, skip_serializing_if = "Option::is_none")] filter: Option>, }, - KeyedSummaryBuild { + /// One keyed state per group whose items take a row weight, e.g. a CMS + /// with heap (built by [`Operator::per_group_keyed_summary_build`]). + PerGroupKeyedSummaryBuild { family: SummaryFamilyType, - value: usize, items: Vec, + /// What each row adds to its item's count. + row_weight: RowWeight, groups: Vec, /// Rows for which this is not true update no state; their group is kept. #[serde(default, skip_serializing_if = "Option::is_none")] filter: Option>, }, - /// One HydraCms state shared by all groups, adding each row's item weight. - SharedSummaryBuild { + /// One state shared by all groups and emitted per group, today a + /// HydraCms grid (built by [`Operator::all_groups_summary_build`]). + AllGroupsSummaryBuild { family: SummaryFamilyType, item: usize, /// What each row adds to its item's count. - row_weight: HydraRowWeight, + row_weight: RowWeight, groups: Vec, /// Rows for which this is not true update no state; their group is kept. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -311,9 +317,9 @@ impl PhysicalOperator for Operator { | Kind::Window { .. } | Kind::Join { .. } | Kind::SemiJoin { .. } - | Kind::SummaryBuild { .. } - | Kind::KeyedSummaryBuild { .. } - | Kind::SharedSummaryBuild { .. } + | Kind::PerGroupSummaryBuild { .. } + | Kind::PerGroupKeyedSummaryBuild { .. } + | Kind::AllGroupsSummaryBuild { .. } | Kind::SummaryMerge { .. } | Kind::VectorToScalar { .. } ) @@ -364,9 +370,9 @@ impl PhysicalOperator for Operator { Kind::Window { .. } => "WindowAggregate", Kind::SemiJoin { .. } => "SemiJoin", Kind::Join { .. } => "RelationalJoin", - Kind::SummaryBuild { .. } - | Kind::KeyedSummaryBuild { .. } - | Kind::SharedSummaryBuild { .. } => "SummaryAgg", + Kind::PerGroupSummaryBuild { .. } + | Kind::PerGroupKeyedSummaryBuild { .. } + | Kind::AllGroupsSummaryBuild { .. } => "SummaryAgg", Kind::KeyedEvaluation { .. } => "SummaryEstimate", Kind::SummaryMerge { .. } => "SummaryMerge", Kind::Evaluation { .. } => "SummaryEvaluation", @@ -418,10 +424,10 @@ impl PhysicalOperator for Operator { } Kind::Join { .. } | Kind::SemiJoin { .. } => joins::execute(self, inputs, context), Kind::SummaryMerge { .. } => summary::execute_merge(self, inputs, context), - Kind::SummaryBuild { .. } + Kind::PerGroupSummaryBuild { .. } | Kind::Evaluation { .. } - | Kind::KeyedSummaryBuild { .. } - | Kind::SharedSummaryBuild { .. } + | Kind::PerGroupKeyedSummaryBuild { .. } + | Kind::AllGroupsSummaryBuild { .. } | Kind::KeyedEvaluation { .. } => summary::execute(self, inputs, context), } } diff --git a/crates/executor/src/operators/summary/mod.rs b/crates/executor/src/operators/summary/mod.rs index 51b9d8db..93209cb7 100644 --- a/crates/executor/src/operators/summary/mod.rs +++ b/crates/executor/src/operators/summary/mod.rs @@ -2,17 +2,18 @@ use super::*; use planner_types::ir::schema::{FamilyCapability, MergeInputRule}; use std::num::NonZeroU32; -/// What each row of a shared HydraCms build adds to its item's count. -/// Produced by the physical planner from a validated `SummaryUpdate` -/// weight and read once per row by the build. +/// What each input row adds to its item's count in a keyed summary build +/// ([`Operator::per_group_keyed_summary_build`] or +/// [`Operator::all_groups_summary_build`]). Produced by the physical planner +/// from a validated `SummaryUpdate.weight` and read once per row. #[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub enum HydraRowWeight { - /// The value of this input column (Float64 or Int64, non-negative); a - /// row whose value is NULL adds nothing. +pub enum RowWeight { + /// The value of this input column (Float64 or Int64); a row whose value + /// is NULL adds nothing. Column(usize), - /// This positive integer per row: 1 for a row count. At most `i32::MAX`, - /// since asap_sketchlib's Hydra adds `i32` weights; - /// [`Operator::shared_summary_build`] refuses a larger one. + /// This positive integer per row: 1 for a row count. The all-groups + /// (HydraCms) build refuses one above `i32::MAX`, since + /// asap_sketchlib's Hydra adds `i32` weights. Constant(NonZeroU32), } /// A summary evaluation: a sketch query, or an exact evaluation with typed parameters. @@ -23,11 +24,15 @@ pub enum SummaryEvaluation { } impl Operator { - pub fn keyed_summary_build( + /// One keyed summary state per group (a CMS or Count Sketch with heap): + /// each row adds its `row_weight` to the tuple of its `items` values, in + /// its group's state. A weight column must be Float64 or Int64, the + /// types the kernel's `f64` weights hold; a NULL weight adds nothing. + pub fn per_group_keyed_summary_build( input: SchemaRef, family: SummaryFamilyType, - value: usize, items: Vec, + row_weight: RowWeight, groups: Vec, ) -> Result { use crate::summary_kernels::weighted_frequency::WeightedFrequency; @@ -37,10 +42,16 @@ impl Operator { }; WeightedFrequency::configuration(kind)?; validate_groups(&input, &groups)?; - if items.is_empty() || plain(&input, value)? != (&DataType::Float64, false) { - return Err(invalid( - "keyed summary requires identities and non-null Float64 weights", - )); + if items.is_empty() { + return Err(invalid("keyed summary requires item columns")); + } + if let RowWeight::Column(weight) = row_weight { + if !matches!( + plain(&input, weight)?.0, + DataType::Float64 | DataType::Int64 + ) { + return Err(invalid("keyed summary weights must be Float64 or Int64")); + } } for &item in &items { if !matches!( @@ -66,10 +77,10 @@ impl Operator { table: None, }); Ok(Self { - kind: Kind::KeyedSummaryBuild { + kind: Kind::PerGroupKeyedSummaryBuild { family, - value, items, + row_weight, groups, filter: None, }, @@ -77,14 +88,16 @@ impl Operator { output: schema(fields), }) } - /// One HydraCms state shared by every group, emitted as one row per group - /// that reads it. Each row adds its `row_weight` to its `item` value; a - /// row whose weight column is NULL adds nothing. - pub fn shared_summary_build( + /// One summary state shared by all groups (a HydraCms grid), emitted as + /// one row per group that reads its own part of it. Each row adds its + /// `row_weight` to its `item` value in its group; a NULL weight adds + /// nothing. A constant weight is at most `i32::MAX` (asap_sketchlib's + /// Hydra adds `i32` weights). + pub fn all_groups_summary_build( input: SchemaRef, family: SummaryFamilyType, item: usize, - row_weight: HydraRowWeight, + row_weight: RowWeight, groups: Vec, ) -> Result { crate::values::validate_family(&family)?; @@ -99,7 +112,7 @@ impl Operator { )); } match row_weight { - HydraRowWeight::Column(weight) => { + RowWeight::Column(weight) => { if !matches!( plain(&input, weight)?.0, DataType::Float64 | DataType::Int64 @@ -107,7 +120,7 @@ impl Operator { return Err(invalid("HydraCms weights must be Float64 or Int64")); } } - HydraRowWeight::Constant(weight) => { + RowWeight::Constant(weight) => { if i32::try_from(weight.get()).is_err() { return Err(Error::Invalid(format!( "HydraCms constant weight must be at most i32::MAX (asap_sketchlib's \ @@ -127,7 +140,7 @@ impl Operator { table: None, }); Ok(Self { - kind: Kind::SharedSummaryBuild { + kind: Kind::AllGroupsSummaryBuild { family, item, row_weight, @@ -146,9 +159,9 @@ impl Operator { return Err(invalid("summary filter must be boolean")); } match &mut self.kind { - Kind::SummaryBuild { filter: slot, .. } - | Kind::KeyedSummaryBuild { filter: slot, .. } - | Kind::SharedSummaryBuild { filter: slot, .. } => *slot = Some(Box::new(filter)), + Kind::PerGroupSummaryBuild { filter: slot, .. } + | Kind::PerGroupKeyedSummaryBuild { filter: slot, .. } + | Kind::AllGroupsSummaryBuild { filter: slot, .. } => *slot = Some(Box::new(filter)), _ => return Err(invalid("a row filter needs a summary build")), } Ok(self) @@ -183,7 +196,11 @@ impl Operator { output, }) } - pub fn summary_build( + /// One summary state per group, updated from the value of one input + /// column per row (`time` orders counter samples): an exact aggregate, + /// a quantile/distinct sketch, or an unkeyed frequency sketch (UnivMon) + /// over that column's values. + pub fn per_group_summary_build( input: SchemaRef, family: SummaryFamilyType, value: usize, @@ -235,7 +252,7 @@ impl Operator { table: None, }); Ok(Self { - kind: Kind::SummaryBuild { + kind: Kind::PerGroupSummaryBuild { family, value, time, @@ -334,7 +351,7 @@ pub(super) fn execute<'a>( let output = operator.output.clone(); let input = inputs.pop().ok_or_else(|| invalid("input missing"))?; match &operator.kind { - Kind::SummaryBuild { + Kind::PerGroupSummaryBuild { family, value, time, @@ -343,25 +360,33 @@ pub(super) fn execute<'a>( } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_summary(input, family, *value, *time, groups, filter.as_deref(), !operator.inputs[0].has_promql_series_identity(), &context).await?, + build_per_group_summary(input, family, *value, *time, groups, filter.as_deref(), !operator.inputs[0].has_promql_series_identity(), &context).await?, ) }) .boxed_local()), - Kind::KeyedSummaryBuild { + Kind::PerGroupKeyedSummaryBuild { family, - value, items, + row_weight, groups, filter, } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_keyed_summary(input, family, *value, items, groups, filter.as_deref(), &context) - .await?, + build_per_group_keyed_summary( + input, + family, + items, + *row_weight, + groups, + filter.as_deref(), + &context, + ) + .await?, ) }) .boxed_local()), - Kind::SharedSummaryBuild { + Kind::AllGroupsSummaryBuild { family, item, row_weight, @@ -370,7 +395,7 @@ pub(super) fn execute<'a>( } => Ok(futures::stream::once(async move { Batch::try_new( output, - build_shared_summary( + build_all_groups_summary( input, family, *item, @@ -507,7 +532,7 @@ pub(super) fn execute_merge<'a>( } #[allow(clippy::too_many_arguments)] -async fn build_summary( +async fn build_per_group_summary( mut input: Input<'_, Batch>, family: &SummaryFamilyType, value: usize, @@ -739,11 +764,11 @@ async fn merge_summary( .collect()) } -async fn build_keyed_summary( +async fn build_per_group_keyed_summary( mut input: Input<'_, Batch>, family: &SummaryFamilyType, - value: usize, items: &[usize], + row_weight: RowWeight, groups: &[usize], filter: Option<&Expression>, context: &RunContext, @@ -785,10 +810,14 @@ async fn build_keyed_summary( if !selected(filter, row)? { continue; } - let (_, summary, reservation, overhead) = states.get_mut(&key).unwrap(); - let Value::Float64(weight) = row[value] else { - return Err(invalid("weighted frequency weight type")); + let weight = match row_weight { + RowWeight::Constant(weight) => f64::from(weight.get()), + RowWeight::Column(i) => match &row[i] { + Value::Null => continue, + value => keyed_weight(value)?, + }, }; + let (_, summary, reservation, overhead) = states.get_mut(&key).unwrap(); summary.update( &items .iter() @@ -814,11 +843,11 @@ async fn build_keyed_summary( .collect()) } -async fn build_shared_summary( +async fn build_all_groups_summary( mut input: Input<'_, Batch>, family: &SummaryFamilyType, item: usize, - row_weight: HydraRowWeight, + row_weight: RowWeight, groups: &[usize], filter: Option<&Expression>, context: &RunContext, @@ -853,10 +882,10 @@ async fn build_shared_summary( continue; } let weight = match row_weight { - HydraRowWeight::Constant(weight) => { + RowWeight::Constant(weight) => { i32::try_from(weight.get()).expect("validated HydraCms constant weight") } - HydraRowWeight::Column(i) => match &row[i] { + RowWeight::Column(i) => match &row[i] { Value::Null => continue, value => hydra_weight(value)?, }, @@ -887,6 +916,19 @@ async fn build_shared_summary( .collect()) } +/// One row's weight-column value as the keyed kernel's `f64` weight. An +/// Int64 weight beyond 2^53 has no exact `f64` and is refused, not rounded. +fn keyed_weight(value: &Value) -> Result { + match value { + Value::Float64(v) => Ok(*v), + Value::Int64(v) if v.unsigned_abs() <= 1 << 53 => Ok(*v as f64), + Value::Int64(v) => Err(Error::Operator(format!( + "keyed summary weight {v} has no exact Float64" + ))), + _ => Err(Error::Operator("keyed summary weight type".into())), + } +} + /// One row's HydraCms weight. sketchlib's Hydra adds `i32` weights, so a /// weight must be a non-negative integer within `i32`. fn hydra_weight(value: &Value) -> Result { diff --git a/crates/executor/src/operators/unchecked.rs b/crates/executor/src/operators/unchecked.rs index 3b40dd7c..3fd9f21e 100644 --- a/crates/executor/src/operators/unchecked.rs +++ b/crates/executor/src/operators/unchecked.rs @@ -146,34 +146,40 @@ impl TryFrom for Operator { &planner_types::ir::Predicate(predicate.expression().clone()), output.clone(), )?, - Kind::SummaryBuild { + Kind::PerGroupSummaryBuild { family, value, time, groups, filter, } => with_filter( - Operator::summary_build(input(0)?, family, value, time, groups)?, + Operator::per_group_summary_build(input(0)?, family, value, time, groups)?, filter, )?, - Kind::KeyedSummaryBuild { + Kind::PerGroupKeyedSummaryBuild { family, - value, items, + row_weight, groups, filter, } => with_filter( - Operator::keyed_summary_build(input(0)?, family, value, items, groups)?, + Operator::per_group_keyed_summary_build( + input(0)?, + family, + items, + row_weight, + groups, + )?, filter, )?, - Kind::SharedSummaryBuild { + Kind::AllGroupsSummaryBuild { family, item, row_weight, groups, filter, } => with_filter( - Operator::shared_summary_build(input(0)?, family, item, row_weight, groups)?, + Operator::all_groups_summary_build(input(0)?, family, item, row_weight, groups)?, filter, )?, Kind::KeyedEvaluation { state, k } => { diff --git a/crates/executor/src/physical_planner/mod.rs b/crates/executor/src/physical_planner/mod.rs index 9a9b5e58..8ee3e9a3 100644 --- a/crates/executor/src/physical_planner/mod.rs +++ b/crates/executor/src/physical_planner/mod.rs @@ -2,7 +2,7 @@ //! Compilation needs no readers; deployment resolves inputs after selection. //! A node without a native lowering fails compilation; the deployment runtime //! must execute it. -use crate::operators::{HydraRowWeight, SummaryEvaluation}; +use crate::operators::{RowWeight, SummaryEvaluation}; use crate::summary_kernels::exact::ExactEvaluation; use crate::{ operators::{Expression, Operator, Reduction, SortKey}, @@ -464,7 +464,7 @@ fn compile_internal( let groups = (0..input.fields.len()) .filter(|&column| column != value && column != coordinate) .collect(); - let build = Operator::summary_build( + let build = Operator::per_group_summary_build( input, family.clone(), value, @@ -971,7 +971,7 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result build.with_row_filter(expression(&filter.0, input)?), None => Ok(build), @@ -1007,9 +1007,10 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result { - HydraRowWeight::Column(named_column(input, weight)?) - } - SummaryInputExpr::Constant(weight) => HydraRowWeight::Constant( - crate::capability::hydra_constant_weight(*weight).map_err(Error::Invalid)?, - ), - _ => unreachable!("validated HydraCms weight"), - }; - return Operator::shared_summary_build( + return Operator::all_groups_summary_build( input.clone(), family.clone(), named_column(input, item)?, - row_weight, + row_weight(input, &update.weight)?, groups(input, keys)?, ); } @@ -1057,7 +1049,7 @@ fn summary_build( "summary construction requires explicit grouping columns", )); }; - return Operator::summary_build( + return Operator::per_group_summary_build( input.clone(), family.clone(), named_column(input, item)?, @@ -1069,11 +1061,6 @@ fn summary_build( let PlannerReduction::Reduce(keys) = reduction else { return Err(invalid("keyed summary requires explicit partitions")); }; - let SummaryInputExpr::Column(weight) = &update.weight else { - return Err(invalid( - "keyed summary weight must be a finalized value column", - )); - }; if matches!(family, FieldDataType::Sketch(kind, _) if kind.algorithm() == &planner_types::ir::schema::SketchAlgorithm::CmsWithHeap) && !matches!( update.weight_domain, @@ -1100,11 +1087,11 @@ fn summary_build( } let mut items = Vec::new(); columns(item, input, &mut items)?; - return Operator::keyed_summary_build( + return Operator::per_group_keyed_summary_build( input.clone(), family.clone(), - named_column(input, weight)?, items, + row_weight(input, &update.weight)?, groups(input, keys)?, ); } @@ -1119,7 +1106,7 @@ fn summary_build( "summary construction requires explicit grouping columns", )); }; - Operator::summary_build( + Operator::per_group_summary_build( input.clone(), family.clone(), named_column(input, column)?, @@ -1127,6 +1114,19 @@ fn summary_build( groups(input, keys)?, ) } +/// What each row adds in a keyed build: a projected weight column, or a +/// positive integer constant (1 for a row count) read without a column. +fn row_weight(input: &SchemaRef, weight: &SummaryInputExpr) -> Result { + match weight { + SummaryInputExpr::Column(weight) => Ok(RowWeight::Column(named_column(input, weight)?)), + SummaryInputExpr::Constant(weight) => Ok(RowWeight::Constant( + crate::capability::constant_row_weight(*weight).map_err(Error::Invalid)?, + )), + _ => Err(invalid( + "keyed summary weight must be a projected column or a constant", + )), + } +} /// Families whose state counts values of one column without keyed items: /// their unit-frequency update builds over that column. fn unkeyed_frequency_family(family: &FieldDataType) -> bool { diff --git a/crates/executor/src/physical_planner/precompute.rs b/crates/executor/src/physical_planner/precompute.rs index a20a2f7e..bd421134 100644 --- a/crates/executor/src/physical_planner/precompute.rs +++ b/crates/executor/src/physical_planner/precompute.rs @@ -493,6 +493,14 @@ fn fragment( without: keys.is_without(), }, }; + // A keyed build adds a constant weight (1 for a row count) + // without projecting it as a column. + let constant_weight = match &update.weight { + SummaryInputExpr::Constant(weight) if keyed => { + Some(crate::capability::constant_row_weight(*weight).map_err(Error::Invalid)?) + } + _ => None, + }; let weight = match &update.weight { _ if unit_frequency => Expression::Column(2), SummaryInputExpr::Constant(value) => Expression::Literal { @@ -521,6 +529,11 @@ fn fragment( let mut fields = population_schema(SummaryFamilyType::Plain(DataType::Float64)) .fields .clone(); + if constant_weight.is_some() { + columns.pop(); + fields.pop(); + } + let first_item = fields.len(); if keyed { let mut items = Vec::new(); raw_items( @@ -539,7 +552,7 @@ fn fragment( columns.push((name, expression)); } } - let item_columns = (3..fields.len()).collect::>(); + let item_columns = (first_item..fields.len()).collect::>(); let project = Operator::project(input.clone(), columns)?.with_output_schema( Arc::new(Schema { fields, @@ -551,9 +564,15 @@ fn fragment( let projected = project.schema(); let project = add(vec![0], project)?; let build = if keyed { - Operator::keyed_summary_build(projected, family.clone(), 2, item_columns, vec![0])? + Operator::per_group_keyed_summary_build( + projected, + family.clone(), + item_columns, + constant_weight.map_or(RowWeight::Column(2), RowWeight::Constant), + vec![0], + )? } else { - Operator::summary_build(projected, family.clone(), 2, Some(1), vec![0])? + Operator::per_group_summary_build(projected, family.clone(), 2, Some(1), vec![0])? }; let built = build.schema(); let build = add(vec![project], build)?; diff --git a/crates/executor/tests/blocking_resources.rs b/crates/executor/tests/blocking_resources.rs index cf7309a3..db1e1c10 100644 --- a/crates/executor/tests/blocking_resources.rs +++ b/crates/executor/tests/blocking_resources.rs @@ -234,7 +234,14 @@ fn weighted_summary_build_yields_within_a_batch() { ), Default::default(), ); - let operator = Operator::keyed_summary_build(input, family, 1, vec![0], vec![]).unwrap(); + let operator = Operator::per_group_keyed_summary_build( + input, + family, + vec![0], + asap_executor::dag::operators::RowWeight::Column(1), + vec![], + ) + .unwrap(); let run = context(16 * 1024 * 1024); let inputs = sources.execute(&[0], run.clone()).unwrap(); let mut output = operator.start(inputs, run.clone()).unwrap(); diff --git a/crates/executor/tests/current_series_heap.rs b/crates/executor/tests/current_series_heap.rs index 0d89edd8..09d1abfa 100644 --- a/crates/executor/tests/current_series_heap.rs +++ b/crates/executor/tests/current_series_heap.rs @@ -171,7 +171,14 @@ fn spatial_heap_ranks_latest_values_in_independent_runs() { }, }; let family = FieldDataType::Sketch(SketchKind::new(algorithm, params), Default::default()); - let build = Operator::keyed_summary_build(schema(), family, 1, vec![3], vec![2]).unwrap(); + let build = Operator::per_group_keyed_summary_build( + schema(), + family, + vec![3], + asap_executor::dag::operators::RowWeight::Column(1), + vec![2], + ) + .unwrap(); let output = Arc::new(planner_types::ir::schema::Schema { unique_keys: vec![], closed: false, diff --git a/crates/executor/tests/hydra_cms_execution.rs b/crates/executor/tests/hydra_cms_execution.rs index 3fe0cbb7..1474bf0d 100644 --- a/crates/executor/tests/hydra_cms_execution.rs +++ b/crates/executor/tests/hydra_cms_execution.rs @@ -2,7 +2,7 @@ //! planner and answers each group's item point counts from one shared grid. use asap_executor::{ dag::{ - operators::{HydraRowWeight, Operator}, + operators::{Operator, RowWeight}, values::{Batch, SchemaRef, Value}, Limits, RunContext, Scope, }, @@ -375,18 +375,18 @@ fn row_weight_serializes_as_one_field_and_rejects_constants_above_i32() { ] { let build = compile_node(&summary_agg(update, None), &[input_schema()]).unwrap(); let wire = serde_json::to_value(&build).unwrap(); - let fields = &wire["kind"]["SharedSummaryBuild"]; + let fields = &wire["kind"]["AllGroupsSummaryBuild"]; assert_eq!(fields["row_weight"], expected, "{wire}"); assert!(fields.get("weight").is_none() && fields.get("constant_weight").is_none()); let restored: Operator = serde_json::from_value(wire.clone()).unwrap(); assert_eq!(serde_json::to_value(&restored).unwrap(), wire); } let build = |weight: u32| { - Operator::shared_summary_build( + Operator::all_groups_summary_build( input_schema(), family(), 1, - HydraRowWeight::Constant(std::num::NonZeroU32::new(weight).unwrap()), + RowWeight::Constant(std::num::NonZeroU32::new(weight).unwrap()), vec![0], ) }; @@ -396,7 +396,7 @@ fn row_weight_serializes_as_one_field_and_rejects_constants_above_i32() { .expect("constant above i32::MAX"); assert!(error.to_string().contains("i32::MAX"), "{error}"); let mut wire = serde_json::to_value(build(2).unwrap()).unwrap(); - wire["kind"]["SharedSummaryBuild"]["row_weight"] = + wire["kind"]["AllGroupsSummaryBuild"]["row_weight"] = serde_json::json!({"Constant": i32::MAX as u32 + 1}); let error = serde_json::from_value::(wire) .err() diff --git a/crates/executor/tests/keyed_summary_row_weight.rs b/crates/executor/tests/keyed_summary_row_weight.rs new file mode 100644 index 00000000..2d253eed --- /dev/null +++ b/crates/executor/tests/keyed_summary_row_weight.rs @@ -0,0 +1,326 @@ +//! A per-group keyed summary build (a CMS or Count Sketch with heap per +//! group) adds each row's `RowWeight` to the row's item: the value of a +//! weight column, or a constant such as 1 for a row count. +use asap_executor::{ + dag::{ + operators::{Operator, RowWeight}, + values::{Batch, SchemaRef, Value}, + Limits, PhysicalDAG, RunContext, Scope, + }, + physical_planner::compile_node, +}; +use futures::{executor::block_on, StreamExt}; +use planner_types::ir::operator::{GroupKeys, Reduction}; +use planner_types::ir::physical_export::{ + PhysicalASAPDAGNode, PhysicalASAPOperatorPayload as Payload, +}; +use planner_types::ir::properties::ExecutionDataState; +use planner_types::ir::scalar::ColumnRef; +use planner_types::ir::schema::*; +use planner_types::ir::ASAPOp; +use planner_types::physical::Materialization; +use std::sync::Arc; + +/// `(job, service)` rows: jobs are the groups, services the items. +const ROWS: [(&str, &str); 6] = [ + ("api", "checkout"), + ("api", "checkout"), + ("api", "search"), + ("batch", "export"), + ("batch", "export"), + ("batch", "export"), +]; + +fn heap_family() -> FieldDataType { + FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::CmsWithHeap, + SketchParams::CmsWithHeap { + width: 1024, + depth: 3, + heap_size: 8, + }, + ), + GroupingStrategy::PerSubpopulationInstance, + ) +} + +fn schema(fields: Vec) -> SchemaRef { + Arc::new(Schema { + fields, + unique_keys: vec![], + closed: false, + time_index: None, + }) +} + +/// `(job, service)` input schema, plus a `weight` column of `weight` type +/// when given. +fn input_schema(weight: Option<(DataType, bool)>) -> SchemaRef { + let mut fields = vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("service", DataType::Utf8, false), + ]; + if let Some((dtype, nullable)) = weight { + fields.push(Field::plain("weight", dtype, nullable)); + } + schema(fields) +} + +/// `ROWS` with each row's `weight` value appended, when given. +fn batch(input: &SchemaRef, weight: Option) -> Batch { + Batch::try_new( + input.clone(), + ROWS.iter() + .map(|(job, service)| { + let mut row = vec![Value::Utf8((*job).into()), Value::Utf8((*service).into())]; + row.extend(weight.clone()); + row + }) + .collect(), + ) + .unwrap() +} + +/// Run `build` over `batch` and read each group's top items as sorted +/// `(job, service, score)` rows. +fn top_items(build: Operator, input: SchemaRef, batch: Batch) -> Vec<(String, String, f64)> { + let output = schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("service", DataType::Utf8, false), + Field::plain("score", DataType::Float64, false), + ]); + let evaluation = Operator::keyed_evaluation(build.schema(), 1, 8, output).unwrap(); + let mut dag = PhysicalDAG::default(); + dag.add(0, vec![], Operator::source(input, vec![batch]).unwrap()) + .unwrap(); + dag.add(1, vec![0], build).unwrap(); + dag.add(2, vec![1], evaluation).unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let mut rows = block_on(async { + let mut rows = Vec::new(); + let mut stream = dag.execute(&[2], context).unwrap().remove(0); + while let Some(batch) = stream.next().await { + rows.extend(batch.unwrap().rows().iter().cloned()); + } + rows + }) + .into_iter() + .map(|row| match (&row[0], &row[1], &row[2]) { + (Value::Utf8(job), Value::Utf8(service), Value::Float64(score)) => { + (job.to_string(), service.to_string(), *score) + } + other => panic!("unexpected row {other:?}"), + }) + .collect::>(); + rows.sort_by(|a, b| (&a.0, &a.1).cmp(&(&b.0, &b.1))); + rows +} + +/// The count of each `(job, service)` in `ROWS`, times `weight`. +fn counts(weight: f64) -> Vec<(String, String, f64)> { + [ + ("api", "checkout", 2.), + ("api", "search", 1.), + ("batch", "export", 3.), + ] + .map(|(job, service, count)| (job.to_string(), service.to_string(), count * weight)) + .to_vec() +} + +// A count-ranked heap (`topk by(job)(k, count by(job, service)(...))`) +// updates by the unit count: `SummaryUpdate.weight = Constant(1.0)`. At query +// time it binds to a keyed build whose row weight is the constant 1, not a +// materialized column of ones. +#[test] +fn query_time_count_heap_binds_a_constant_row_weight() { + let input = input_schema(None); + let node = PhysicalASAPDAGNode { + id: 1, + payload: Payload::ASAP(ASAPOp::SummaryAgg { + child: 0, + family: heap_family(), + input: SummaryUpdate { + item: Some(SummaryInputExpr::Tuple(vec![SummaryInputExpr::Column( + ColumnRef::Named("service".into()), + )])), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }, + reduction: Reduction::Reduce(GroupKeys::by(vec![0])), + grouping: GroupingStrategy::PerSubpopulationInstance, + filter: None, + }), + output_state: ExecutionDataState::QUERY_ROWS, + materialization: Materialization::NotMaterialized, + output_schema: (*schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field { + name: "state".into(), + dtype: heap_family(), + nullable: false, + table: None, + }, + ])) + .clone(), + guarantee: None, + }; + let build = compile_node(&node, std::slice::from_ref(&input)).unwrap(); + let wire = serde_json::to_value(&build).unwrap(); + assert_eq!( + wire["kind"]["PerGroupKeyedSummaryBuild"]["row_weight"], + serde_json::json!({"Constant": 1}), + "{wire}" + ); + assert_eq!( + top_items(build, input.clone(), batch(&input, None)), + counts(1.) + ); +} + +/// A keyed build of `ROWS` by `job`, items `service`, adding `row_weight`. +fn keyed_build(input: &SchemaRef, row_weight: RowWeight) -> Operator { + Operator::per_group_keyed_summary_build( + input.clone(), + heap_family(), + vec![1], + row_weight, + vec![0], + ) + .unwrap() +} + +// A constant row weight adds the same as a column holding that constant: +// `Constant(3)` over `ROWS` equals `Column(2)` over a Float64 column of 3.0, +// and both are 3 x each item's row count. +#[test] +fn constant_row_weight_equals_a_column_of_that_constant() { + let plain = input_schema(None); + let constant = top_items( + keyed_build( + &plain, + RowWeight::Constant(std::num::NonZeroU32::new(3).unwrap()), + ), + plain.clone(), + batch(&plain, None), + ); + let weighted = input_schema(Some((DataType::Float64, false))); + let column = top_items( + keyed_build(&weighted, RowWeight::Column(2)), + weighted.clone(), + batch(&weighted, Some(Value::Float64(3.))), + ); + assert_eq!(constant, column); + assert_eq!(constant, counts(3.)); +} + +// An Int64 weight column adds the same as a Float64 column of equal values; +// an Int64 weight beyond 2^53 (no exact Float64) is refused, not rounded. +#[test] +fn int64_weight_column_equals_float64_column() { + let int64 = input_schema(Some((DataType::Int64, false))); + assert_eq!( + top_items( + keyed_build(&int64, RowWeight::Column(2)), + int64.clone(), + batch(&int64, Some(Value::Int64(2))), + ), + counts(2.) + ); + let build = keyed_build(&int64, RowWeight::Column(2)); + let evaluation = Operator::keyed_evaluation( + build.schema(), + 1, + 8, + schema(vec![ + Field::plain("job", DataType::Utf8, false), + Field::plain("service", DataType::Utf8, false), + Field::plain("score", DataType::Float64, false), + ]), + ) + .unwrap(); + let mut dag = PhysicalDAG::default(); + dag.add( + 0, + vec![], + Operator::source( + int64.clone(), + vec![batch(&int64, Some(Value::Int64((1 << 53) + 1)))], + ) + .unwrap(), + ) + .unwrap(); + dag.add(1, vec![0], build).unwrap(); + dag.add(2, vec![1], evaluation).unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let error = block_on(dag.execute(&[2], context).unwrap().remove(0).next()) + .unwrap() + .err() + .expect("Int64 weight beyond 2^53 accepted"); + assert!(error.to_string().contains("no exact Float64"), "{error}"); + // A Utf8 weight column is refused when the build is constructed. + let text = input_schema(Some((DataType::Utf8, false))); + let error = Operator::per_group_keyed_summary_build( + text, + heap_family(), + vec![1], + RowWeight::Column(2), + vec![0], + ) + .err() + .expect("Utf8 weight accepted"); + assert!(error.to_string().contains("Float64 or Int64"), "{error}"); +} + +// A NULL in a nullable weight column adds nothing, as with SQL `SUM`: no +// item is counted, while each row's group still gets a state. +#[test] +fn null_weight_adds_nothing() { + let nullable = input_schema(Some((DataType::Float64, true))); + let nulls = batch(&nullable, Some(Value::Null)); + let mut dag = PhysicalDAG::default(); + dag.add( + 0, + vec![], + Operator::source(nullable.clone(), vec![nulls.clone()]).unwrap(), + ) + .unwrap(); + dag.add(1, vec![0], keyed_build(&nullable, RowWeight::Column(2))) + .unwrap(); + let context = RunContext::new( + Scope::Query { + evaluation_time_ms: 0, + revision: 1, + }, + Limits::default(), + ) + .unwrap(); + let states = block_on(dag.execute(&[1], context).unwrap().remove(0).next()) + .unwrap() + .unwrap(); + assert_eq!(states.rows().len(), 2, "one state per job"); + assert_eq!( + top_items( + keyed_build(&nullable, RowWeight::Column(2)), + nullable, + nulls + ), + vec![] + ); +} diff --git a/crates/executor/tests/physical_dag.rs b/crates/executor/tests/physical_dag.rs index 0513fbb1..132b3a40 100644 --- a/crates/executor/tests/physical_dag.rs +++ b/crates/executor/tests/physical_dag.rs @@ -128,7 +128,7 @@ fn summary_construction_merge_and_evaluation_at_both_phases() { .map(|v| Batch::try_new(schema.clone(), vec![vec![Value::Float64(v as f64)]]).unwrap()) .collect(); let family = FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum); - let build = Operator::summary_build(schema.clone(), family, 0, None, vec![]).unwrap(); + let build = Operator::per_group_summary_build(schema.clone(), family, 0, None, vec![]).unwrap(); let state = build.schema(); let mut dag = PhysicalDAG::default(); dag.add(0, vec![], Operator::source(schema, batches).unwrap()) @@ -285,7 +285,7 @@ fn scalar_negation_and_vector_conversion() { #[test] fn binding_rejects_unsupported_operations() { let schema = schema(&[("v", DataType::Float64, false)]); - assert!(Operator::summary_build( + assert!(Operator::per_group_summary_build( schema.clone(), FieldDataType::ExactAggregate(ExactKind::Rate, ExactParams::Rate), 0, @@ -293,7 +293,7 @@ fn binding_rejects_unsupported_operations() { vec![] ) .is_err()); - let sum = Operator::summary_build( + let sum = Operator::per_group_summary_build( schema.clone(), FieldDataType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), 0, @@ -328,7 +328,7 @@ fn kll_raw_partial_and_precomputed_are_native_dags() { ), GroupingStrategy::PerSubpopulationInstance, ); - let build = Operator::summary_build(input.clone(), family, 0, None, vec![]).unwrap(); + let build = Operator::per_group_summary_build(input.clone(), family, 0, None, vec![]).unwrap(); let state = build.schema(); let build_range = |start: u32, end: u32| { let mut dag = PhysicalDAG::default(); @@ -581,7 +581,7 @@ fn bind_post_asap_before_execution() { #[test] fn empty_exact_count_is_an_integer_state_evaluation() { let input = schema(&[("value", DataType::Float64, false)]); - let build = Operator::summary_build( + let build = Operator::per_group_summary_build( input.clone(), FieldDataType::ExactAggregate(ExactKind::Count, ExactParams::Count), 0, @@ -1110,7 +1110,14 @@ fn assert_weighted_rate_topk(count_sketch: bool) { ), Default::default(), ); - let build = Operator::keyed_summary_build(rates.schema(), family, 3, vec![0], vec![1]).unwrap(); + let build = Operator::per_group_keyed_summary_build( + rates.schema(), + family, + vec![0], + asap_executor::dag::operators::RowWeight::Column(3), + vec![1], + ) + .unwrap(); let output = schema(&[ ("job", DataType::Utf8, false), ("service", DataType::Utf8, false), diff --git a/crates/executor/tests/physical_semantics.rs b/crates/executor/tests/physical_semantics.rs index 178c7f1a..b42fe266 100644 --- a/crates/executor/tests/physical_semantics.rs +++ b/crates/executor/tests/physical_semantics.rs @@ -583,11 +583,12 @@ fn kll_partial_merge_and_multiple_evaluations_preserve_population() { dag.add( id + 3, vec![id], - Operator::summary_build(input.clone(), family.clone(), 0, None, vec![]).unwrap(), + Operator::per_group_summary_build(input.clone(), family.clone(), 0, None, vec![]) + .unwrap(), ) .unwrap(); } - let state = Operator::summary_build(input, family, 0, None, vec![]) + let state = Operator::per_group_summary_build(input, family, 0, None, vec![]) .unwrap() .schema(); dag.add(6, vec![3, 4], Operator::union(state.clone(), 2).unwrap()) @@ -668,7 +669,9 @@ fn compacted_kll_panes_merge_within_the_calibrated_rank_error() { ), Default::default(), ); - let build = || Operator::summary_build(input.clone(), family.clone(), 0, None, vec![]).unwrap(); + let build = || { + Operator::per_group_summary_build(input.clone(), family.clone(), 0, None, vec![]).unwrap() + }; let state = build().schema(); let mut dag = PhysicalDAG::default(); for pane in 0..PANES { @@ -741,8 +744,10 @@ fn counter_panes_merge_in_any_order() { ("t", DataType::Timestamp, false), ]); let family = FieldDataType::ExactAggregate(ExactKind::Increase, ExactParams::Increase); - let build = - || Operator::summary_build(input.clone(), family.clone(), 0, Some(1), vec![]).unwrap(); + let build = || { + Operator::per_group_summary_build(input.clone(), family.clone(), 0, Some(1), vec![]) + .unwrap() + }; let state = build().schema(); let samples = |range: std::ops::Range| -> Vec> { range @@ -829,7 +834,7 @@ fn empty_exact_summary_extrema_agree_with_ordinary_aggregation() { (ExactKind::Min, ExactParams::Min, Statistic::Min), (ExactKind::Max, ExactParams::Max, Statistic::Max), ] { - let build = Operator::summary_build( + let build = Operator::per_group_summary_build( input.clone(), FieldDataType::ExactAggregate(kind, params), 0, diff --git a/crates/executor/tests/univmon_execution.rs b/crates/executor/tests/univmon_execution.rs index 3eed50ef..996f7b59 100644 --- a/crates/executor/tests/univmon_execution.rs +++ b/crates/executor/tests/univmon_execution.rs @@ -105,7 +105,7 @@ fn one_univmon_state_answers_distinct_l2_and_entropy() -> Result<(), Error> { .map(|value| vec![Value::Float64(value)]) .collect(), )?; - let build = Operator::summary_build(input, family(), 0, None, vec![])?; + let build = Operator::per_group_summary_build(input, family(), 0, None, vec![])?; let [distinct, l2, entropy] = three_statistics(batch, build)?[..] else { unreachable!() }; @@ -148,7 +148,7 @@ fn typed_frequency_keys_preserve_identity() -> Result<(), Error> { vec![Value::Null], ], )?; - let build = Operator::summary_build(input, family(), 0, None, vec![])?; + let build = Operator::per_group_summary_build(input, family(), 0, None, vec![])?; for (found, expected) in three_statistics(batch, build)? .into_iter() diff --git a/crates/executor/tests/weighted_topk_binding.rs b/crates/executor/tests/weighted_topk_binding.rs index 7a69de72..a0f271f5 100644 --- a/crates/executor/tests/weighted_topk_binding.rs +++ b/crates/executor/tests/weighted_topk_binding.rs @@ -269,7 +269,7 @@ fn direct_rate_topk_preserves_dynamic_unreferenced_labels() { )); assert_eq!(ranked.input_contracts().count(), 1); let encoded = String::from_utf8(serde_json::to_vec(&ranked).unwrap()).unwrap(); - assert!(encoded.contains("KeyedSummaryBuild")); + assert!(encoded.contains("PerGroupKeyedSummaryBuild")); assert!(encoded.contains("KeyedEvaluation")); assert!( !encoded.contains("\"Rate\""), @@ -736,12 +736,12 @@ fn maintained_rate_heap_compiles_fixed_window_precompute() { } let precompute = String::from_utf8(serde_json::to_vec(&physical.precompute.unwrap()).unwrap()).unwrap(); - assert!(precompute.contains("KeyedSummaryBuild")); + assert!(precompute.contains("PerGroupKeyedSummaryBuild")); assert!(precompute.contains("Rate")); assert!( !String::from_utf8(serde_json::to_vec(&physical.query).unwrap()) .unwrap() - .contains("KeyedSummaryBuild") + .contains("PerGroupKeyedSummaryBuild") ); } } diff --git a/crates/integration-tests/tests/kll_pane_execution.rs b/crates/integration-tests/tests/kll_pane_execution.rs index e4b5c819..501394da 100644 --- a/crates/integration-tests/tests/kll_pane_execution.rs +++ b/crates/integration-tests/tests/kll_pane_execution.rs @@ -55,7 +55,8 @@ fn query_scope() -> Scope { } fn fixture() -> (CompiledPhysicalDAG, Operator, SchemaRef) { let raw = raw_schema(); - let build = Operator::summary_build(raw.clone(), family(200), 0, None, vec![]).unwrap(); + let build = + Operator::per_group_summary_build(raw.clone(), family(200), 0, None, vec![]).unwrap(); let state = build.schema(); let maintenance = CompiledPhysicalDAG::from_operators( BTreeMap::from([(0, InputContract::bounded(raw))]), diff --git a/crates/integration-tests/tests/precompute_raw_samples.rs b/crates/integration-tests/tests/precompute_raw_samples.rs index ccfe1f58..b5e045d7 100644 --- a/crates/integration-tests/tests/precompute_raw_samples.rs +++ b/crates/integration-tests/tests/precompute_raw_samples.rs @@ -566,8 +566,20 @@ fn raw_sample_heaps_resolve_items_from_labels() { (count_sketch.clone(), identity), (count_sketch, by_instance.clone()), ] { + let unit_count = input.weight == SummaryInputExpr::Constant(1.0); let (dag, source, root) = grouped_raw_summary(family, input); assert!(check("heap", &dag, source, root, &rows).is_ok()); + // A unit count adds the constant 1 per row; no column of ones is + // projected for it. + let compiled = + serde_json::to_value(precompute::compile(&dag, &[source], &[root]).unwrap()).unwrap(); + assert_eq!( + compiled + .to_string() + .contains(r#""row_weight":{"Constant":1}"#), + unit_count, + "{compiled}" + ); } let signed_cms = grouped_raw_summary(cms.clone(), by_instance.clone()); assert!(precompute::compile(&signed_cms.0, &[signed_cms.1], &[signed_cms.2]).is_err()); @@ -631,3 +643,47 @@ fn raw_sample_without_grouping_drops_labels_and_name() { Ok("Sum".into()) ); } + +// End to end: `topk(2, count by (service) (m))` at ε = 0.02 has a Count-Min +// heap candidate whose update is the unit count (`Constant(1.0)`). Placed at +// ingestion, its raw-sample precompute adds the constant 1 per sample to each +// service's count, with no column of ones projected, and each heap reads the +// exact per-service sample counts of the fixture. +#[test] +fn count_ranked_heap_precompute_adds_a_constant_row_weight() { + let query = "topk(2, count by (service) (m))"; + let rows = samples(); + let mut checked = 0; + for candidate in candidates(query, AccuracyTarget::max_error(0.02)) { + let dag = compile_at_ingestion_physical_asap_dag(&candidate).unwrap(); + for (source, root) in raw_summaries(&dag) { + let node = dag.nodes.iter().find(|n| n.id as u64 == root).unwrap(); + let PhysicalASAPOperatorPayload::ASAP(ASAPOp::SummaryAgg { + family: FieldDataType::Sketch(kind, _), + input, + .. + }) = &node.payload + else { + continue; + }; + if kind.algorithm() != &SketchAlgorithm::CmsWithHeap { + continue; + } + assert_eq!(input.weight, SummaryInputExpr::Constant(1.0)); + let compiled = + serde_json::to_value(precompute::compile(&dag, &[source], &[root]).unwrap()) + .unwrap() + .to_string(); + assert!( + compiled.contains(r#""row_weight":{"Constant":1}"#), + "{compiled}" + ); + assert_eq!( + check(query, &dag, source, root, &rows), + Ok("CmsWithHeap".into()) + ); + checked += 1; + } + } + assert!(checked > 0, "{query} has no Count-Min heap candidate"); +} From 48ac3019c4b7cc1a0f19daf12f3fff32d93018b0 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Sun, 18 Oct 2026 20:29:02 +0000 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20refuse=20a=20HydraCms=20constant=20w?= =?UTF-8?q?eight=20above=20i32::MAX=20at=20validation;=20test=20the=20?= =?UTF-8?q?=C2=B12^53=20Int64=20weight=20edge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit validate_hydra_cms now applies the i32::MAX bound after constant_row_weight, so the kernel check and the all-groups constructor agree. Reword the no-kernel error for a shared-grouping (Hydra) family. --- crates/executor/src/capability.rs | 12 ++++-- crates/executor/tests/hydra_cms_execution.rs | 26 ++++++++++++ .../tests/keyed_summary_row_weight.rs | 41 +++++++++++++++++++ 3 files changed, 75 insertions(+), 4 deletions(-) diff --git a/crates/executor/src/capability.rs b/crates/executor/src/capability.rs index d6127eb0..b0925666 100644 --- a/crates/executor/src/capability.rs +++ b/crates/executor/src/capability.rs @@ -148,9 +148,13 @@ fn validate_hydra_cms( match (&input.weight, proof) { (SummaryInputExpr::Constant(w), NonNegativeWeightProof::UnitCount) if *w == 1.0 => Ok(()), (SummaryInputExpr::Constant(w), NonNegativeWeightProof::PositiveConstant) => { - constant_row_weight(*w) - .map(|_| ()) - .map_err(|error| format!("HydraCms {error}")) + let weight = constant_row_weight(*w).map_err(|error| format!("HydraCms {error}"))?; + i32::try_from(weight.get()).map(|_| ()).map_err(|_| { + format!( + "HydraCms constant weight must be at most i32::MAX (asap_sketchlib's \ + Hydra adds i32 weights), got {weight}" + ) + }) } (SummaryInputExpr::Column(_), _) => Ok(()), _ => Err( @@ -197,7 +201,7 @@ pub(crate) fn hydra_cms_shape( }, ) = family else { - return Err("shared summary grouping has no registered kernel".into()); + return Err("a shared-grouping (Hydra) family has no registered kernel".into()); }; if kind.algorithm() != &SketchAlgorithm::Cms || kind.params() diff --git a/crates/executor/tests/hydra_cms_execution.rs b/crates/executor/tests/hydra_cms_execution.rs index 1474bf0d..4897a99d 100644 --- a/crates/executor/tests/hydra_cms_execution.rs +++ b/crates/executor/tests/hydra_cms_execution.rs @@ -404,6 +404,32 @@ fn row_weight_serializes_as_one_field_and_rejects_constants_above_i32() { assert!(error.to_string().contains("i32::MAX"), "{error}"); } +// A constant weight above `i32::MAX` (asap_sketchlib's Hydra adds `i32` +// weights) is refused by the kernel check, not only by the constructor. +#[test] +fn constant_weight_above_i32_is_refused_at_validation() { + let update = SummaryUpdate { + weight: SummaryInputExpr::Constant(2_147_483_648.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::PositiveConstant, + }, + ..unit_count() + }; + let error = asap_executor::capability::validate_summary_kernel(&family(), &update, &grouping()) + .expect_err("validation accepted a constant above i32::MAX"); + assert!( + error.contains("HydraCms") && error.contains("i32::MAX"), + "{error}" + ); + let error = compile_node(&summary_agg(update, None), &[input_schema()]) + .err() + .expect("constant above i32::MAX compiled"); + assert!( + error.to_string().contains("HydraCms") && error.to_string().contains("i32::MAX"), + "{error}" + ); +} + // A weight column of unknown sign is rejected at binding: a Count-Min point // count only overestimates when no weight is negative. #[test] diff --git a/crates/executor/tests/keyed_summary_row_weight.rs b/crates/executor/tests/keyed_summary_row_weight.rs index 2d253eed..1c10e393 100644 --- a/crates/executor/tests/keyed_summary_row_weight.rs +++ b/crates/executor/tests/keyed_summary_row_weight.rs @@ -236,6 +236,47 @@ fn int64_weight_column_equals_float64_column() { ), counts(2.) ); + // ±2^53 is the largest Int64 magnitude with an exact Float64, so it + // builds: 2^53 on the Count-Min heap (non-negative weights only), -2^53 + // on the Count Sketch heap, which takes signed weights. + let limit = 1i64 << 53; + assert_eq!( + top_items( + keyed_build(&int64, RowWeight::Column(2)), + int64.clone(), + batch(&int64, Some(Value::Int64(limit))), + ), + counts(limit as f64) + ); + let count_sketch = FieldDataType::Sketch( + SketchKind::new( + SketchAlgorithm::CountSketchWithHeap, + SketchParams::CountSketchWithHeap { + width: 1024, + depth: 3, + heap_size: 8, + }, + ), + GroupingStrategy::PerSubpopulationInstance, + ); + let signed = Operator::per_group_keyed_summary_build( + int64.clone(), + count_sketch, + vec![1], + RowWeight::Column(2), + vec![0], + ) + .unwrap(); + assert_eq!( + top_items( + signed, + int64.clone(), + batch(&int64, Some(Value::Int64(-limit))) + ) + .len(), + 3, + "-2^53 builds and every item is read back" + ); let build = keyed_build(&int64, RowWeight::Column(2)); let evaluation = Operator::keyed_evaluation( build.schema(),