diff --git a/Cargo.lock b/Cargo.lock index af73a82c..dd1e13c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -340,6 +340,26 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "asap-ir" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6" +dependencies = [ + "serde", + "serde_json", + "thiserror 2.0.18", +] + +[[package]] +name = "asap-ir" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=d4c175633f6ce46b801ca2115c00fa757d5b6240#d4c175633f6ce46b801ca2115c00fa757d5b6240" +dependencies = [ + "serde", + "serde_json", + "thiserror 2.0.18", +] + [[package]] name = "asap-ir" version = "0.1.0" @@ -353,19 +373,20 @@ dependencies = [ [[package]] name = "asap-l2" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4#fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6" dependencies = [ - "asap-ir", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", "thiserror 2.0.18", ] [[package]] name = "asap-plan" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4#fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6" dependencies = [ - "asap-ir", - "asap-sketch", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", + "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", + "serde_json", "thiserror 2.0.18", ] @@ -379,12 +400,28 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "asap-sketch" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6#64df20d90c3ddd519c726dea45c05e1fe5225ce6" +dependencies = [ + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", +] + +[[package]] +name = "asap-sketch" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPController?rev=d4c175633f6ce46b801ca2115c00fa757d5b6240#d4c175633f6ce46b801ca2115c00fa757d5b6240" +dependencies = [ + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=d4c175633f6ce46b801ca2115c00fa757d5b6240)", +] + [[package]] name = "asap-sketch" version = "0.1.0" source = "git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4#fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" dependencies = [ - "asap-ir", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)", ] [[package]] @@ -408,6 +445,7 @@ dependencies = [ "rmp-serde", "serde", "serde-big-array", + "serde_bytes", "smallvec", "twox-hash", "xxhash-rust", @@ -418,8 +456,8 @@ name = "asap_types" version = "0.1.0" dependencies = [ "anyhow", - "asap-ir", - "asap-sketch", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)", + "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4)", "clap 4.6.1", "serde", "serde_json", @@ -789,10 +827,10 @@ name = "control_plane" version = "0.1.0" dependencies = [ "anyhow", - "asap-ir", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", "asap-l2", "asap-plan", - "asap-sketch", + "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=64df20d90c3ddd519c726dea45c05e1fe5225ce6)", "asap_types", "axum", "bytes", @@ -987,9 +1025,9 @@ dependencies = [ "anyhow", "arc-swap", "arrow", - "asap-ir", + "asap-ir 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=d4c175633f6ce46b801ca2115c00fa757d5b6240)", "asap-precompute-rs", - "asap-sketch", + "asap-sketch 0.1.0 (git+https://github.com/ProjectASAP/ASAPController?rev=d4c175633f6ce46b801ca2115c00fa757d5b6240)", "asap_otel_proto", "asap_sketchlib", "asap_types", @@ -1092,7 +1130,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -1797,7 +1835,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi 0.5.2", "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2979,7 +3017,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3110,6 +3148,16 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_bytes" +version = "0.11.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8" +dependencies = [ + "serde", + "serde_core", +] + [[package]] name = "serde_core" version = "1.0.228" @@ -3410,7 +3458,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -4211,7 +4259,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 1aec834c..764594ee 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -61,10 +61,17 @@ asap_types.workspace = true # this doesn't pull in datafusion or any front-end weight. asap-sketch is # asap-plan's own dependency (SummaryKind/SummaryParams), needed here only # to translate Implementation into this repo's own Capability vocabulary. -asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" } -asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" } -asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" } -asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "fc09c3aa0b0cf0297ee415d1ed49f7cdc0cc55e4" } +# +# Bumped to 64df20d (main tip, "feat(plan): pluggable AggIntent::Extension +# realization + readout hook", #162) to pick up +# `CostModel::{realize_extension, readout_extension}` -- the pluggable +# `AggIntent::Extension` hook this repo's `Extension{"frequency"}` intent +# needs (ASAPController#150). 64df20d is a strict descendant of d4c1756 +# (the previous pin), so nothing this repo already consumes moves. +asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" } +asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" } +asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" } +asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "64df20d90c3ddd519c726dea45c05e1fe5225ce6" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 5967661a..fc8330cb 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -1024,10 +1024,13 @@ mod runtime_tests { Some(BTreeSet::from([SummaryKind::CountSketchWithHeap])), ), // `CountMinSketch` override re-derives statistic to - // `Frequency`, `AggIntent::Extension`-shaped — declines to - // bind pending ASAPController#150 (see - // `optimizer::rules::tests::typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support`). - ("endpoint_request_freq", None), + // `Frequency`, `AggIntent::Extension`-shaped — now binds via + // `ControlPlaneCostModel::realize_extension` (ASAPController#150, + // see `optimizer::rules::tests::typed_binding_endpoint_request_freq_binds_cms`). + ( + "endpoint_request_freq", + Some(BTreeSet::from([SummaryKind::Cms])), + ), ]; for (metric, want) in &expected { let got = map.get(*metric).cloned(); @@ -1037,12 +1040,12 @@ mod runtime_tests { full map: {map:?}", ); } - // Routing table covers the 4 sketched metrics (endpoint_request_freq - // and http_requests_total both decline — see above). + // Routing table covers the 5 sketched metrics (only + // http_requests_total declines, as raw passthrough). assert_eq!( map.len(), - 4, - "routing table should have 4 entries (4 sketches; raw + Extension both decline), got: {map:?}" + 5, + "routing table should have 5 entries (5 sketches; only raw passthrough declines), got: {map:?}" ); } @@ -1225,7 +1228,11 @@ mod runtime_tests { store.set( METRIC, AggRole::Other, - mk(AggType::Frequency, Some(SketchType::CountSketch), Vec::new()), + mk( + AggType::Frequency, + Some(SketchType::CountSketch), + Vec::new(), + ), WorkloadCharacteristics::default(), ); diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 17838170..68ebcd98 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -3038,9 +3038,10 @@ fn build_backend_readout_json(r: &BackendReadout) -> JsonValue { SketchQuery::Cardinality => json!({ "op": "cardinality", }), - SketchQuery::PointCount { key } => json!({ + SketchQuery::PointCount { key, value } => json!({ "op": "point_count", "key": column_ref_to_wire_key(key), + "value": value, }), SketchQuery::TopK { k } => json!({ "op": "topk", @@ -3613,6 +3614,7 @@ mod tests { aggregation_id: "agg1".into(), op: SketchQuery::PointCount { key: ColumnRef::Named("user_42".into()), + value: None, }, }, ], @@ -3700,6 +3702,7 @@ mod tests { SummaryKind::CountSketch => SketchQuery::TopK { k: 10 }, SummaryKind::Cms => SketchQuery::PointCount { key: ColumnRef::Named("user_42".into()), + value: None, }, other => unreachable!( "backend_cfg_with_kind: unsupported test fixture kind {other:?}" diff --git a/control_plane/src/intent_algebra/agg_intent.rs b/control_plane/src/intent_algebra/agg_intent.rs index 902c15f3..b40f9989 100644 --- a/control_plane/src/intent_algebra/agg_intent.rs +++ b/control_plane/src/intent_algebra/agg_intent.rs @@ -57,7 +57,13 @@ pub use asap_ir::intent_algebra::{ use crate::intent_algebra::schema::{Column, DataType}; use crate::types_v2::AccuracyTarget; -const FREQUENCY_EXT_KIND: &str = "frequency"; +/// `ext_kind` tag for control_plane's point-frequency-via-CMS intent. +/// `pub(crate)` (not just module-private) so +/// `sketch_algebra::cost_model::ControlPlaneCostModel`'s +/// `realize_extension`/`readout_extension` can match on it directly -- +/// those take `(ext_kind: &str, payload: &serde_json::Value)`, not a +/// whole `AggIntent`, so they can't call [`as_frequency`] itself. +pub(crate) const FREQUENCY_EXT_KIND: &str = "frequency"; /// Construct control_plane's point-frequency-via-CMS intent. See module /// docs for why this is an `Extension`, not a shared first-class variant. diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index 7bb2db28..4b747598 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -259,12 +259,13 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option SummaryKind::CountSketchWithHeap, SummaryKind::Cms if statistic == StatisticClass::TopK => SummaryKind::CmsWithHeap, @@ -278,12 +279,13 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option)`, just wrapping the input as `SummaryExpr::Logical` - // unchanged. `bind_workload_typed`'s own contract is `None` for - // "typed path doesn't support this shape yet" — translate the two - // by checking whether anything actually got committed. + // intent `boundary::implementation_for`/`CostModel::realize_extension` + // can't realize (e.g. `TopK { accuracy: Exact }`, ASAPController#151, + // still open) still returns `Ok(Rc)`, just wrapping the input + // as `SummaryExpr::Logical` unchanged. `bind_workload_typed`'s own + // contract is `None` for "typed path doesn't support this shape yet" + // — translate the two by checking whether anything actually got + // committed. if matches!(node.expr, asap_sketch::SummaryExpr::Logical(_)) { return None; } @@ -505,18 +507,19 @@ mod tests { // `CountMinSketch` re-derives statistic to `Frequency`, which is // `AggIntent::Extension`-shaped (this deployment's point-frequency - // query) — `asap_plan::boundary::implementation_for` maps every - // `Extension` to `PassThrough` unconditionally, regardless of - // which `CostModel` is plugged in (ASAPController#150), so this - // override can no longer bind at all pending the upstream gap - // closing. Same accepted regression as the plain - // `endpoint_request_freq` contract row (see - // `typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support`). + // query). `ControlPlaneCostModel::realize_extension`/ + // `readout_extension` (ASAPController#150) now realize it as + // `SummaryKind::Cms` — matching `capability_matching::pick_family`'s + // own `Frequency -> Cms` mapping — so this override binds like any + // other now. let mut w = workload(vec![AggType::Quantile]); w.sketch_type_override = Some(SketchType::CountMinSketch); - assert!( - bind_workload_typed(&w).is_none(), - "CountMinSketch override should decline pending ASAPController#150", + let pe = bind_workload_typed(&w) + .unwrap_or_else(|| panic!("CountMinSketch override should bind (ASAPController#150)")); + assert_eq!( + extract_root_sketch_kind(&pe), + Some(SummaryKind::Cms), + "CountMinSketch override should pin Cms (Frequency's family), not decline", ); } @@ -740,22 +743,21 @@ mod tests { } #[test] - fn typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support() { - // Contract used to be: `endpoint_request_freq` → CMS (Frequency). - // `Frequency` is `AggIntent::Extension`-shaped; - // `asap_plan::boundary::implementation_for` maps every `Extension` - // to `PassThrough` unconditionally (core has no realization - // opinion for a deployment-specific shape it doesn't know — see - // ASAPController#150). Step B of the plan-shaped-serving - // migration adopted `implement_tree_in_with` here without a local - // Frequency-binding workaround, so this contract row now declines - // the typed path (falls back to the legacy `plan()` output) until - // #150 lands an extension point. + fn typed_binding_endpoint_request_freq_binds_cms() { + // Contract: `endpoint_request_freq` → CMS (Frequency). `Frequency` + // is `AggIntent::Extension`-shaped; this used to decline the typed + // path entirely (`asap_plan::boundary::implementation_for` mapped + // every `Extension` to `PassThrough` unconditionally — core has no + // realization opinion for a deployment-specific shape it doesn't + // know, ASAPController#150). Now that + // `ControlPlaneCostModel::realize_extension`/`readout_extension` + // handle `"frequency"`, this contract row binds like any other. let w = workload_for("endpoint_request_freq", AggType::Frequency); - let bound = bind_workload_typed(&w); - assert!( - bound.is_none(), - "endpoint_request_freq should decline pending ASAPController#150; got {bound:?}", + let bound = bind_workload_typed(&w).expect("endpoint_request_freq must bind"); + assert_eq!( + extract_family(&bound), + Some(SummaryKind::Cms), + "endpoint_request_freq should bind to Cms (Frequency)", ); } @@ -867,9 +869,14 @@ mod tests { AggType::Frequency, Some(SummaryKind::CountSketchWithHeap), ), - // `Extension`/Frequency PassThrough gap — ASAPController#150, - // see `typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support`. - ("endpoint_request_freq", AggType::Frequency, None), + // `Extension`/Frequency now binds via `ControlPlaneCostModel`'s + // `realize_extension` (ASAPController#150) — see + // `typed_binding_endpoint_request_freq_binds_cms`. + ( + "endpoint_request_freq", + AggType::Frequency, + Some(SummaryKind::Cms), + ), ]; for (metric, agg, expected) in cases { let w = workload_for(metric, agg); diff --git a/control_plane/src/sketch_algebra/cost_model.rs b/control_plane/src/sketch_algebra/cost_model.rs index 89bae492..8610c062 100644 --- a/control_plane/src/sketch_algebra/cost_model.rs +++ b/control_plane/src/sketch_algebra/cost_model.rs @@ -13,29 +13,33 @@ //! walk itself (schema derivation, `col`/`by` computation, DAG //! construction) can be `asap_plan::bind`'s rather than a forked copy. //! -//! Two shapes `implement_tree_in_with` cannot realize even with this -//! `CostModel` plugged in, because the decision of *whether* to call into -//! `rank_candidates`/`size_params` at all is made upstream, before the -//! `CostModel` is ever consulted: +//! `AggIntent::Extension` (control_plane's `Frequency` point-query) used +//! to be one of two shapes `implement_tree_in_with` couldn't realize even +//! with this `CostModel` plugged in — `boundary::implementation_for_with` +//! now consults [`ControlPlaneCostModel::realize_extension`]/ +//! [`readout_extension`](CostModel::readout_extension) for it instead of +//! hardcoding `PassThrough` (ASAPController#150). +//! +//! One shape remains genuinely unreachable via this `CostModel`, because +//! the decision of *whether* to call into `rank_candidates`/`size_params` +//! at all is made upstream, before the `CostModel` is ever consulted: //! -//! - `AggIntent::Extension` (control_plane's `Frequency` point-query) — -//! `boundary::implementation_for` maps every `Extension` to -//! `PassThrough` unconditionally, by design (core has no realization -//! opinion for a deployment-specific shape it doesn't know). //! - `AggIntent::TopK { accuracy: AccuracyTarget::Exact, .. }` — routes to //! `exact_realization`, which has no accumulator form for `TopK` and -//! also returns `PassThrough`, even though control_plane's own +//! returns `PassThrough`, even though control_plane's own //! `BindCountSketchOnTopK` still binds this shape (Tight recall tier → -//! `CountSketchWithHeap`). -//! -//! Both are intercepted in `lower.rs` *before* `implement_tree_in_with` -//! runs — see that module's `bind_recursive` for the pre-pass. +//! `CountSketchWithHeap`). Still intercepted in `lower.rs` *before* +//! `implement_tree_in_with` runs — see that module's `bind_recursive` +//! for the pre-pass (ASAPController#151, still open). #![allow(dead_code)] +use asap_ir::intent_algebra::expr_ir::ColumnRef; +use asap_plan::boundary::Implementation; use asap_plan::CostModel; -use asap_sketch::{SummaryKind, SummaryParams}; +use asap_sketch::{SketchQuery, SummaryKind, SummaryParams}; +use crate::intent_algebra::agg_intent::FREQUENCY_EXT_KIND; use crate::intent_algebra::AggIntent; use crate::optimizer::cost::wire::WireCostTable; use crate::types_v2::AccuracyTarget; @@ -64,17 +68,19 @@ impl ControlPlaneCostModel { match (&self.workload_accuracy, intent_accuracy) { (AccuracyTarget::Exact, _) | (_, AccuracyTarget::Exact) => None, (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => Some((a.min(*b), 0.01)), + (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { epsilon, delta }) + | (AccuracyTarget::EpsilonDelta { epsilon, delta }, AccuracyTarget::Epsilon(a)) => { + Some((a.min(*epsilon), *delta)) + } ( - AccuracyTarget::Epsilon(a), - AccuracyTarget::EpsilonDelta { epsilon, delta }, - ) - | ( - AccuracyTarget::EpsilonDelta { epsilon, delta }, - AccuracyTarget::Epsilon(a), - ) => Some((a.min(*epsilon), *delta)), - ( - AccuracyTarget::EpsilonDelta { epsilon: a, delta: da }, - AccuracyTarget::EpsilonDelta { epsilon: b, delta: db }, + AccuracyTarget::EpsilonDelta { + epsilon: a, + delta: da, + }, + AccuracyTarget::EpsilonDelta { + epsilon: b, + delta: db, + }, ) => Some((a.min(*b), da.min(*db))), } } @@ -95,17 +101,19 @@ impl ControlPlaneCostModel { AccuracyTarget::Exact => (0.01, 0.01), }, (AccuracyTarget::Epsilon(a), AccuracyTarget::Epsilon(b)) => (a.min(*b), 0.01), + (AccuracyTarget::Epsilon(a), AccuracyTarget::EpsilonDelta { epsilon, delta }) + | (AccuracyTarget::EpsilonDelta { epsilon, delta }, AccuracyTarget::Epsilon(a)) => { + (a.min(*epsilon), *delta) + } ( - AccuracyTarget::Epsilon(a), - AccuracyTarget::EpsilonDelta { epsilon, delta }, - ) - | ( - AccuracyTarget::EpsilonDelta { epsilon, delta }, - AccuracyTarget::Epsilon(a), - ) => (a.min(*epsilon), *delta), - ( - AccuracyTarget::EpsilonDelta { epsilon: a, delta: da }, - AccuracyTarget::EpsilonDelta { epsilon: b, delta: db }, + AccuracyTarget::EpsilonDelta { + epsilon: a, + delta: da, + }, + AccuracyTarget::EpsilonDelta { + epsilon: b, + delta: db, + }, ) => (a.min(*b), da.min(*db)), } } @@ -237,6 +245,79 @@ impl CostModel for ControlPlaneCostModel { } } } + + /// Realize control_plane's `Frequency` (`ext_kind: "frequency"`) + /// point-query intent (ASAPController#150). `SummaryKind::Cms` (heap- + /// less — a point lookup needs no heap, unlike `TopK`) matches + /// `capability_matching::pick_family`'s own `Frequency → Cms` mapping + /// (and its `is_valid_pair` truth table, which declares `(Cms, + /// Frequency)` valid and has no `(CountSketch, Frequency)` entry) — + /// the newer, tested, currently-authoritative source of truth for this + /// choice, not the older `sketch_catalog::sketch_type_for_op`'s + /// `SketchType::CountSketch` (a genuinely different sketch algorithm + /// under a same-ish name — `SketchType` has separate `CountSketch` + /// and `CountMinSketch` variants; that mapping predates + /// `capability_matching` and disagrees with it). Sized the same + /// `e/eps` width / `ln(1/delta)` depth way every other CMS-family kind + /// here is. `PassThrough` for any other `ext_kind` (none exist yet) + /// or an unparseable/`Exact` accuracy, matching every other + /// approximate-capable intent's `Exact ⇒ no sketch form` policy. + fn realize_extension(&self, ext_kind: &str, payload: &serde_json::Value) -> Implementation { + if ext_kind != FREQUENCY_EXT_KIND { + return Implementation::PassThrough; + } + let Some(accuracy) = payload + .get("accuracy") + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + else { + return Implementation::PassThrough; + }; + let Some((eps, delta)) = self.combined_eps_delta(&accuracy) else { + return Implementation::PassThrough; + }; + let (width, depth) = Self::cms_width_depth(eps, delta); + Implementation::Sketch { + kind: SummaryKind::Cms, + params: SummaryParams::Cms { + width: width.next_power_of_two(), + depth, + }, + } + } + + /// Build the `SketchQuery` readout for `Frequency`. `item_label`/ + /// `item_value` are populated by `intent_algebra::agg_intent::frequency` + /// once the filter value is threaded through (ASAPQuery-backend Phase + /// 3 — not yet); until then `payload` never has them, so this + /// correctly falls back to the bare bucket total + /// (`key: SampleValue, value: None`) — the same answer a `Frequency` + /// intent with no item filter should give either way. + fn readout_extension( + &self, + ext_kind: &str, + payload: &serde_json::Value, + _col: &ColumnRef, + ) -> SketchQuery { + debug_assert_eq!( + ext_kind, FREQUENCY_EXT_KIND, + "readout_extension called for an ext_kind realize_extension never realizes as Sketch" + ); + let item_label = payload.get("item_label").and_then(|v| v.as_str()); + let item_value = payload + .get("item_value") + .and_then(|v| v.as_str()) + .map(str::to_string); + match item_label { + Some(label) => SketchQuery::PointCount { + key: ColumnRef::Named(label.to_string()), + value: item_value, + }, + None => SketchQuery::PointCount { + key: ColumnRef::SampleValue, + value: None, + }, + } + } } /// A `CostModel` that forces a single family for whichever intent it's @@ -260,7 +341,11 @@ impl ForcedFamilyCostModel { } impl CostModel for ForcedFamilyCostModel { - fn rank_candidates(&self, _intent: &AggIntent, _candidates: &[SummaryKind]) -> Vec { + fn rank_candidates( + &self, + _intent: &AggIntent, + _candidates: &[SummaryKind], + ) -> Vec { vec![self.forced.clone()] } @@ -273,6 +358,25 @@ impl CostModel for ForcedFamilyCostModel { ) -> SummaryParams { self.inner.size_params(kind, intent, eps, delta) } + + // `realize_extension`/`readout_extension` delegate to `inner` rather + // than falling back to the trait's default `PassThrough` — otherwise + // `bind_workload_typed`'s `Frequency` contract row (which binds via + // `ForcedFamilyCostModel`, already knowing its family pick from the + // capability matrix) would still decline pending #150 even after + // `ControlPlaneCostModel` itself learned to realize it. + fn realize_extension(&self, ext_kind: &str, payload: &serde_json::Value) -> Implementation { + self.inner.realize_extension(ext_kind, payload) + } + + fn readout_extension( + &self, + ext_kind: &str, + payload: &serde_json::Value, + col: &ColumnRef, + ) -> SketchQuery { + self.inner.readout_extension(ext_kind, payload, col) + } } /// Map an ε rank-error budget to a KLL stream-size `k`. Verbatim port of @@ -325,7 +429,10 @@ mod tests { #[test] fn quantile_always_prefers_ddsketch_over_kll() { let model = ControlPlaneCostModel::new(AccuracyTarget::Epsilon(0.1)); - let ranked = model.rank_candidates(&default_quantile(0.99), &[SummaryKind::Kll, SummaryKind::DDSketch]); + let ranked = model.rank_candidates( + &default_quantile(0.99), + &[SummaryKind::Kll, SummaryKind::DDSketch], + ); assert_eq!(ranked, vec![SummaryKind::DDSketch, SummaryKind::Kll]); } @@ -359,7 +466,10 @@ mod tests { k: 5, accuracy: eps(0.01), }; - let ranked = model.rank_candidates(&intent, &[SummaryKind::CmsWithHeap, SummaryKind::CountSketchWithHeap]); + let ranked = model.rank_candidates( + &intent, + &[SummaryKind::CmsWithHeap, SummaryKind::CountSketchWithHeap], + ); assert_eq!(ranked, vec![SummaryKind::CountSketchWithHeap]); } @@ -370,7 +480,10 @@ mod tests { k: 5, accuracy: eps(0.01), }; - let ranked = model.rank_candidates(&intent, &[SummaryKind::CmsWithHeap, SummaryKind::CountSketchWithHeap]); + let ranked = model.rank_candidates( + &intent, + &[SummaryKind::CmsWithHeap, SummaryKind::CountSketchWithHeap], + ); assert_eq!(ranked[0], SummaryKind::CmsWithHeap); } @@ -382,7 +495,10 @@ mod tests { accuracy: eps(0.01), }; let params = model.size_params(SummaryKind::CmsWithHeap, &intent, 0.0, 0.0); - let SummaryParams::CmsWithHeap { width, heap_size, .. } = params else { + let SummaryParams::CmsWithHeap { + width, heap_size, .. + } = params + else { panic!("expected CmsWithHeap params"); }; assert!(width.is_power_of_two()); diff --git a/control_plane/src/sketch_algebra/lower.rs b/control_plane/src/sketch_algebra/lower.rs index bb98bb09..ec8cb310 100644 --- a/control_plane/src/sketch_algebra/lower.rs +++ b/control_plane/src/sketch_algebra/lower.rs @@ -16,13 +16,18 @@ //! `CostModel` hook can reach, since the decision of *whether* to call //! into `rank_candidates`/`size_params` at all is made before the //! `CostModel` is ever consulted. See each helper's docs for the specific -//! reason. Everything else — including `AggIntent::Extension` (the -//! `Frequency` point-query) and `AggIntent::TopK { accuracy: Exact }`, -//! both of which `implementation_for` maps to `PassThrough` — is left to +//! reason. +//! +//! `AggIntent::Extension` (the `Frequency` point-query) needs no such +//! pre-pass anymore: `ControlPlaneCostModel::realize_extension`/ +//! `readout_extension` (ASAPController#150) now realize it as a real +//! `CountSketch`, so the catch-all arm below commits it via +//! `implement_tree_in_with` like any other intent. +//! `AggIntent::TopK { accuracy: Exact }` is the one remaining case left to //! fall through to `implement_tree_in_with`'s own `Logical` fallback -//! unchanged: both are genuine `asap-plan` coverage gaps, not something -//! this deployment can or should route around locally (filed upstream — -//! see ASAPController#150, #151). +//! unchanged — a genuine, still-open `asap-plan` coverage gap (filed +//! upstream — see ASAPController#151), not something this deployment +//! should route around locally. #![allow(dead_code)] diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs index 347ce035..e0b7a851 100644 --- a/control_plane/src/sketch_algebra/tests.rs +++ b/control_plane/src/sketch_algebra/tests.rs @@ -5,10 +5,13 @@ use std::rc::Rc; use std::time::Duration; +use asap_ir::intent_algebra::expr_ir::ColumnRef; use asap_sketch::{L4Node, SketchQuery, SummaryExpr, SummaryKind, SummaryParams}; use crate::intent_algebra::schema::{Column, DataType}; -use crate::intent_algebra::{AggIntent, BindingScope, LabelFilter, QueryExpr, Schema, Source, WindowKind}; +use crate::intent_algebra::{ + AggIntent, BindingScope, LabelFilter, QueryExpr, Schema, Source, WindowKind, +}; use crate::sketch_algebra::cost_model::ForcedFamilyCostModel; use crate::sketch_algebra::lower::bind_query_expr; use crate::sketch_algebra::physical_expr::{L4Plan, PhysicalExpr}; @@ -126,12 +129,9 @@ fn bind_kll_quantile_basic() { // `bind_picks_ddsketch_over_kll_when_eps_explicit` below. let expr = agg_quantile(0.99, AccuracyTarget::Epsilon(0.01)); let cost_model = ForcedFamilyCostModel::new(AccuracyTarget::Epsilon(0.01), SummaryKind::Kll); - let node = asap_plan::bind::implement_tree_in_with( - &expr, - &BindingScope::default(), - &cost_model, - ) - .expect("KLL should bind a Quantile{0.99, ε=0.01}"); + let node = + asap_plan::bind::implement_tree_in_with(&expr, &BindingScope::default(), &cost_model) + .expect("KLL should bind a Quantile{0.99, ε=0.01}"); match &node.expr { SummaryExpr::SummaryEstimate { query, @@ -163,12 +163,9 @@ fn bind_ddsketch_quantile_basic() { let expr = agg_quantile(0.99, AccuracyTarget::Epsilon(0.01)); let cost_model = ForcedFamilyCostModel::new(AccuracyTarget::Epsilon(0.01), SummaryKind::DDSketch); - let node = asap_plan::bind::implement_tree_in_with( - &expr, - &BindingScope::default(), - &cost_model, - ) - .expect("DDSketch should bind a Quantile{0.99, ε=0.01}"); + let node = + asap_plan::bind::implement_tree_in_with(&expr, &BindingScope::default(), &cost_model) + .expect("DDSketch should bind a Quantile{0.99, ε=0.01}"); match &node.expr { SummaryExpr::SummaryEstimate { query, @@ -176,9 +173,7 @@ fn bind_ddsketch_quantile_basic() { } => { assert!(matches!(query, SketchQuery::Quantile { q } if *q == 0.99)); match &sketch_input.expr { - SummaryExpr::SummaryAgg { - sketch, params, .. - } => { + SummaryExpr::SummaryAgg { sketch, params, .. } => { assert_eq!(sketch, &SummaryKind::DDSketch); match params { SummaryParams::DDSketch { alpha } => { @@ -246,9 +241,7 @@ fn topk_binding_family(bound: &PhysicalExpr) -> (SummaryKind, u32, u32) { } => { assert!(matches!(query, SketchQuery::TopK { k } if *k == 10)); match &sketch_input.expr { - SummaryExpr::SummaryAgg { - sketch, params, .. - } => match params { + SummaryExpr::SummaryAgg { sketch, params, .. } => match params { SummaryParams::CmsWithHeap { width, depth, .. } => { (sketch.clone(), *width, *depth) } @@ -312,8 +305,8 @@ fn bind_cms_topk_tight_recall_picks_countsketch() { // Intent requests a normal (non-exact) rank so binding still // happens; the workload-level policy demands exact recall. let expr = agg_topk(10, AccuracyTarget::Epsilon(0.01)); - let bound = bind_query_expr(&expr, AccuracyTarget::Exact) - .expect("bind_query_expr should not error"); + let bound = + bind_query_expr(&expr, AccuracyTarget::Exact).expect("bind_query_expr should not error"); let (kind, w, d) = topk_binding_family(&bound); assert_eq!( kind, @@ -380,9 +373,7 @@ fn bind_hll_cardinality_basic() { } => { assert!(matches!(query, SketchQuery::Cardinality)); match &sketch_input.expr { - SummaryExpr::SummaryAgg { - sketch, params, .. - } => { + SummaryExpr::SummaryAgg { sketch, params, .. } => { assert_eq!(sketch, &SummaryKind::Hll); match params { SummaryParams::Hll { precision } => { @@ -425,7 +416,11 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() { match bound { PhysicalExpr::Committed(L4Plan::Summary(node)) => match &node.expr { SummaryExpr::SummaryAgg { sketch, params, .. } => { - assert_eq!(sketch, &SummaryKind::Sum, "Sum should bind to SummaryAgg(Sum)"); + assert_eq!( + sketch, + &SummaryKind::Sum, + "Sum should bind to SummaryAgg(Sum)" + ); assert_eq!(params, &SummaryParams::Sum); } other => panic!("expected bare SummaryAgg(Sum), got {other:?}"), @@ -487,10 +482,7 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() { assert!(matches!(query, SketchQuery::Quantile { .. })); match &sketch_input.expr { SummaryExpr::SummaryAgg { sketch, .. } => { - assert!(matches!( - sketch, - SummaryKind::Kll | SummaryKind::DDSketch - )); + assert!(matches!(sketch, SummaryKind::Kll | SummaryKind::DDSketch)); } other => panic!("expected SummaryAgg under SummaryEstimate, got {other:?}"), } @@ -868,18 +860,22 @@ fn phase_b_archive_only_intents_round_trip_through_binder() { // ── Deliberate behavior changes (ASAPController#150 / #151) ────────────────── // // `AggIntent::Extension` (this deployment's `Frequency` point-query, -// built via `crate::intent_algebra::frequency(accuracy)`) and -// `AggIntent::TopK { accuracy: Exact }` both now decline to bind +// built via `crate::intent_algebra::frequency(accuracy)`) now binds to a +// real `Cms` sketch via `ControlPlaneCostModel::realize_extension`/ +// `readout_extension` (ASAPController#150) — see `frequency_extension_binds_cms` +// below and `optimizer::rules::mod::tests::typed_binding_endpoint_request_freq_binds_cms`. +// `AggIntent::TopK { accuracy: Exact }` still declines to bind // (`SummaryExpr::Logical`) rather than sketch — a REAL, accepted -// behavior change introduced by this migration (core's -// `boundary::implementation_for` maps every `Extension` to -// `PassThrough` unconditionally, and `TopK{Exact}`'s `exact_realization` -// has no accumulator form for it either — see `lower.rs`'s module docs -// and `cost_model.rs`'s module docs). Mirrors the pattern already used in -// `optimizer::rules::mod::tests::typed_binding_endpoint_request_freq_declines_pending_upstream_extension_support`. +// behavior change from this migration that remains open +// (`TopK{Exact}`'s `exact_realization` has no accumulator form for it — +// see `lower.rs`'s module docs and `cost_model.rs`'s module docs, +// ASAPController#151, still open). #[test] -fn frequency_extension_declines_pending_upstream_extension_support() { +fn frequency_extension_binds_cms() { + // `ControlPlaneCostModel::realize_extension`/`readout_extension` + // (ASAPController#150) now realize `AggIntent::Extension{"frequency"}` + // as a real `Cms` sketch instead of declining to `Logical`. let intent = crate::intent_algebra::frequency(AccuracyTarget::Epsilon(0.01)); let expr = QueryExpr::Aggregate { by: vec![].into(), @@ -891,10 +887,34 @@ fn frequency_extension_declines_pending_upstream_extension_support() { let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).expect("no error"); match bound { PhysicalExpr::Committed(L4Plan::Summary(node)) => { + let SummaryExpr::SummaryEstimate { + sketch_input, + query, + } = &node.expr + else { + panic!("expected SummaryEstimate, got {:?}", node.expr); + }; assert!( - matches!(&node.expr, SummaryExpr::Logical(_)), - "Frequency (AggIntent::Extension) should decline pending ASAPController#150, got {:?}", - node.expr + matches!( + &sketch_input.expr, + SummaryExpr::SummaryAgg { + sketch: SummaryKind::Cms, + .. + } + ), + "expected a Cms SummaryAgg, got {:?}", + sketch_input.expr + ); + assert!( + matches!( + query, + SketchQuery::PointCount { + key: ColumnRef::SampleValue, + value: None + } + ), + "no filter value threaded through yet (ASAPQuery-backend Phase 3) -- \ + should read out as the bare bucket total, got {query:?}", ); } other => panic!("expected Committed(Summary(_)), got {other:?}"),