Skip to content

Commit 938cfcd

Browse files
zzylolclaude
andcommitted
feat(query-engine): add DDSketch aggregation type
Add a DDSketch quantile aggregation backed by asap_sketchlib::DDSketch, the implementation sketch-bench measures for the planner's cost table. - AggregationType::DDSketch (aliases: DDSketchAccumulator, ddsketch, dd), single-population, served for Statistic::Quantile by capability matching. - DDSketchAccumulator: update, quantile query, merge, msgpack/JSON output. Non-positive values are dropped, as DDSketch indexes positive values only. - Accumulator factory updater; config validation requires an `alpha` parameter in (0, 1). - Planner: DDSketch parameter building (default alpha 0.01, overridable via sketch_parameters.DDSketch.alpha) and sketch properties. The MILP optimizer skips DDSketch until it has a parameter grid and cost rows. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 6888b19 commit 938cfcd

13 files changed

Lines changed: 663 additions & 6 deletions

File tree

‎asap-common/dependencies/rs/asap_types/src/aggregation_config.rs‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,13 @@ pub enum AggregationConfigError {
4040
aggregation_id: u64,
4141
aggregation_type: AggregationType,
4242
},
43+
#[error(
44+
"aggregation {aggregation_id} (DDSketch) parameter 'alpha' must be a number in (0, 1), got {value:?}"
45+
)]
46+
InvalidAlpha {
47+
aggregation_id: u64,
48+
value: Option<Value>,
49+
},
4350
}
4451

4552
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -85,6 +92,7 @@ impl AggregationConfig {
8592
pub fn validate(&self) -> Result<(), AggregationConfigError> {
8693
self.mode().map(|_| ())?;
8794
self.validate_hll_precision()?;
95+
self.validate_ddsketch_alpha()?;
8896
Ok(())
8997
}
9098

@@ -101,6 +109,22 @@ impl AggregationConfig {
101109
)
102110
}
103111

112+
/// DDSketch needs a relative-accuracy `alpha` strictly between 0 and 1;
113+
/// `asap_sketchlib::DDSketch::new` panics on anything else.
114+
fn validate_ddsketch_alpha(&self) -> Result<(), AggregationConfigError> {
115+
if self.aggregation_type != AggregationType::DDSketch {
116+
return Ok(());
117+
}
118+
let value = self.parameters.get("alpha");
119+
match value.and_then(Value::as_f64) {
120+
Some(alpha) if alpha > 0.0 && alpha < 1.0 => Ok(()),
121+
_ => Err(AggregationConfigError::InvalidAlpha {
122+
aggregation_id: self.aggregation_id,
123+
value: value.cloned(),
124+
}),
125+
}
126+
}
127+
104128
fn validate_hll_precision(&self) -> Result<(), AggregationConfigError> {
105129
if self.aggregation_type == AggregationType::HLL {
106130
match self.parameters.get("precision") {

‎asap-common/dependencies/rs/asap_types/src/capability_matching.rs‎

Lines changed: 29 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,11 @@ pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] {
3939
Statistic::Min | Statistic::Max => {
4040
&[AggregationType::MinMax, AggregationType::MultipleMinMax]
4141
}
42-
Statistic::Quantile => &[AggregationType::DatasketchesKLL, AggregationType::HydraKLL],
42+
Statistic::Quantile => &[
43+
AggregationType::DatasketchesKLL,
44+
AggregationType::DDSketch,
45+
AggregationType::HydraKLL,
46+
],
4347
Statistic::Rate | Statistic::Increase => {
4448
&[AggregationType::Increase, AggregationType::MultipleIncrease]
4549
}
@@ -474,6 +478,11 @@ mod tests {
474478
.parameters
475479
.insert("precision".to_string(), serde_json::Value::from(14));
476480
}
481+
if config.aggregation_type == AggregationType::DDSketch {
482+
config
483+
.parameters
484+
.insert("alpha".to_string(), serde_json::Value::from(0.01));
485+
}
477486
config
478487
}
479488

@@ -678,6 +687,25 @@ mod tests {
678687
assert_eq!(result.unwrap().aggregation_id_for_value, 3);
679688
}
680689

690+
#[test]
691+
fn quantile_matches_ddsketch() {
692+
let configs = single_config(make_config(
693+
4,
694+
"lat",
695+
"DDSketch",
696+
"",
697+
300_000,
698+
"tumbling",
699+
&[],
700+
"",
701+
));
702+
let result = find_compatible_aggregation(
703+
&configs,
704+
&req("lat", &[Statistic::Quantile], 300_000, &[], ""),
705+
);
706+
assert_eq!(result.unwrap().aggregation_id_for_value, 4);
707+
}
708+
681709
#[test]
682710
fn no_match_wrong_metric() {
683711
let configs = single_config(make_config(

‎asap-common/dependencies/rs/asap_types/src/streaming_config.rs‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,7 @@ impl Default for StreamingConfig {
136136
mod tests {
137137
use super::*;
138138
use crate::aggregation_config::AggregationConfigError;
139+
use promql_utilities::query_logics::enums::AggregationType;
139140

140141
#[test]
141142
fn rejects_heap_config_with_invalid_sub_type() {
@@ -365,4 +366,60 @@ aggregations:
365366
StreamingConfig::from_yaml_data(&yaml, None)
366367
.expect("MinMax config with 'MAX' subtype must be accepted");
367368
}
369+
370+
fn ddsketch_yaml(parameters_yaml: &str) -> Value {
371+
serde_yaml::from_str(&format!(
372+
r#"
373+
aggregations:
374+
- aggregationId: 1
375+
aggregationType: DDSketch
376+
aggregationSubType: ''
377+
parameters:
378+
{parameters_yaml}
379+
labels:
380+
grouping: [label_0]
381+
aggregated: []
382+
rollup: [instance]
383+
metric: data
384+
windowSizeMs: 60000
385+
slideIntervalMs: 60000
386+
windowType: tumbling
387+
spatialFilter: ''
388+
"#
389+
))
390+
.unwrap()
391+
}
392+
393+
#[test]
394+
fn accepts_ddsketch_config_with_alpha() {
395+
let config = StreamingConfig::from_yaml_data(&ddsketch_yaml("alpha: 0.01"), None)
396+
.expect("DDSketch config with alpha in (0, 1) must load");
397+
let agg = config.get_aggregation_config(1).unwrap();
398+
assert_eq!(agg.aggregation_type, AggregationType::DDSketch);
399+
assert_eq!(agg.parameters["alpha"], serde_json::json!(0.01));
400+
}
401+
402+
#[test]
403+
fn rejects_ddsketch_config_with_missing_or_out_of_range_alpha() {
404+
for parameters in [
405+
"{}",
406+
"alpha: 0",
407+
"alpha: 1",
408+
"alpha: -0.1",
409+
r#"alpha: "0.01""#,
410+
] {
411+
let error = StreamingConfig::from_yaml_data(&ddsketch_yaml(parameters), None)
412+
.expect_err("DDSketch config without a valid alpha must be rejected");
413+
assert!(
414+
matches!(
415+
error.downcast_ref::<AggregationConfigError>(),
416+
Some(AggregationConfigError::InvalidAlpha {
417+
aggregation_id: 1,
418+
..
419+
})
420+
),
421+
"parameters {parameters}: unexpected error {error}"
422+
);
423+
}
424+
}
368425
}

‎asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -278,6 +278,7 @@ pub enum AggregationType {
278278
Increase,
279279
MinMax,
280280
DatasketchesKLL,
281+
DDSketch,
281282
// ---------- multi-population (keyed) ----------
282283
MultipleSum,
283284
MultipleIncrease,
@@ -298,6 +299,7 @@ impl AggregationType {
298299
AggregationType::Increase => "Increase",
299300
AggregationType::MinMax => "MinMax",
300301
AggregationType::DatasketchesKLL => "DatasketchesKLL",
302+
AggregationType::DDSketch => "DDSketch",
301303
AggregationType::MultipleSum => "MultipleSum",
302304
AggregationType::MultipleIncrease => "MultipleIncrease",
303305
AggregationType::MultipleMinMax => "MultipleMinMax",
@@ -362,6 +364,7 @@ impl FromStr for AggregationType {
362364
"Increase" => Ok(AggregationType::Increase),
363365
"MinMax" => Ok(AggregationType::MinMax),
364366
"DatasketchesKLL" => Ok(AggregationType::DatasketchesKLL),
367+
"DDSketch" => Ok(AggregationType::DDSketch),
365368
"MultipleSum" => Ok(AggregationType::MultipleSum),
366369
"MultipleIncrease" => Ok(AggregationType::MultipleIncrease),
367370
"MultipleMinMax" => Ok(AggregationType::MultipleMinMax),
@@ -380,6 +383,7 @@ impl FromStr for AggregationType {
380383
"DatasketchesKLLAccumulator" | "KLL" | "kll" | "datasketches_kll" => {
381384
Ok(AggregationType::DatasketchesKLL)
382385
}
386+
"DDSketchAccumulator" | "ddsketch" | "dd" => Ok(AggregationType::DDSketch),
383387
"MultipleSumAccumulator" | "multiple_sum" => Ok(AggregationType::MultipleSum),
384388
"MultipleIncreaseAccumulator" | "multiple_increase" => {
385389
Ok(AggregationType::MultipleIncrease)

‎asap-common/dependencies/rs/promql_utilities/src/query_logics/logics.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,8 @@ pub fn does_precompute_operator_support_subpopulations(
6868
AggregationType::Increase
6969
| AggregationType::MinMax
7070
| AggregationType::Sum
71-
| AggregationType::DatasketchesKLL => false,
71+
| AggregationType::DatasketchesKLL
72+
| AggregationType::DDSketch => false,
7273

7374
// Multi-key operators
7475
AggregationType::MultipleIncrease

‎asap-planner-rs/src/config/input.rs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -173,6 +173,8 @@ pub struct SketchParameterOverrides {
173173
pub count_min_sketch_with_heap: Option<CmsHeapParams>,
174174
#[serde(rename = "DatasketchesKLL")]
175175
pub datasketches_kll: Option<KllParams>,
176+
#[serde(rename = "DDSketch")]
177+
pub ddsketch: Option<DDSketchParams>,
176178
#[serde(rename = "HydraKLL")]
177179
pub hydra_kll: Option<HydraParams>,
178180
#[serde(rename = "HLL")]
@@ -198,6 +200,11 @@ pub struct KllParams {
198200
pub k: u64,
199201
}
200202

203+
#[derive(Debug, Clone, Deserialize)]
204+
pub struct DDSketchParams {
205+
pub alpha: f64,
206+
}
207+
201208
#[derive(Debug, Clone, Deserialize)]
202209
pub struct HydraParams {
203210
pub row_num: u64,

‎asap-planner-rs/src/optimizer/candidate_gen.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ const OPTIMIZER_SKIPPED_AGG_TYPES: &[AggregationType] = &[
2525
AggregationType::Sum,
2626
AggregationType::MinMax,
2727
AggregationType::Increase,
28+
// No parameter grid or cost rows yet; the optimizer side lands with #762.
29+
AggregationType::DDSketch,
2830
];
2931

3032
/// A candidate streaming config for one optimizer item, ready for cost evaluation.

‎asap-planner-rs/src/optimizer/sketch_properties.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ pub fn sketch_properties(t: AggregationType) -> SketchProperties {
2323
AggregationType::Increase => p(true, false, false),
2424
AggregationType::MinMax => p(true, false, false),
2525
AggregationType::DatasketchesKLL => p(true, false, false),
26+
// Bucket counts could be subtracted, but the accumulator only merges.
27+
AggregationType::DDSketch => p(true, false, false),
2628
AggregationType::MultipleSum => p(true, true, true),
2729
AggregationType::MultipleIncrease => p(true, false, true),
2830
AggregationType::MultipleMinMax => p(true, false, true),

‎asap-planner-rs/src/planner/sketch.rs‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ const DEFAULT_CMS_DEPTH: u64 = 3;
99
const DEFAULT_CMS_WIDTH: u64 = 1024;
1010
const DEFAULT_CMS_HEAP_MULT: u64 = 4;
1111
const DEFAULT_KLL_K: u64 = 500;
12+
const DEFAULT_DDSKETCH_ALPHA: f64 = 0.01;
1213
const DEFAULT_HYDRA_ROW: u64 = 3;
1314
const DEFAULT_HYDRA_COL: u64 = 1024;
1415
const DEFAULT_HYDRA_K: u64 = 20;
@@ -98,6 +99,16 @@ pub fn build_sketch_parameters(
9899
Ok(m)
99100
}
100101

102+
AggregationType::DDSketch => {
103+
let alpha = sketch_params
104+
.and_then(|p| p.ddsketch.as_ref())
105+
.map(|p| p.alpha)
106+
.unwrap_or(DEFAULT_DDSKETCH_ALPHA);
107+
let mut m = HashMap::new();
108+
m.insert("alpha".to_string(), serde_json::json!(alpha));
109+
Ok(m)
110+
}
111+
101112
AggregationType::HLL => {
102113
let precision = sketch_params
103114
.and_then(|p| p.hll.as_ref())
@@ -175,3 +186,29 @@ pub fn build_sketch_parameters_from_promql(
175186
sketch_params,
176187
)
177188
}
189+
190+
#[cfg(test)]
191+
mod tests {
192+
use super::*;
193+
use crate::config::input::DDSketchParams;
194+
195+
#[test]
196+
fn ddsketch_uses_default_alpha_without_override() {
197+
let params =
198+
build_sketch_parameters(AggregationType::DDSketch, "", None, None, None).unwrap();
199+
assert_eq!(params.len(), 1);
200+
assert_eq!(params["alpha"], serde_json::json!(DEFAULT_DDSKETCH_ALPHA));
201+
}
202+
203+
#[test]
204+
fn ddsketch_alpha_override_is_applied() {
205+
let overrides = SketchParameterOverrides {
206+
ddsketch: Some(DDSketchParams { alpha: 0.02 }),
207+
..Default::default()
208+
};
209+
let params =
210+
build_sketch_parameters(AggregationType::DDSketch, "", None, None, Some(&overrides))
211+
.unwrap();
212+
assert_eq!(params["alpha"], serde_json::json!(0.02));
213+
}
214+
}

0 commit comments

Comments
 (0)