feat(control_plane): Phase 2 steps 3-9 -- merge L2/L3 onto asap-ir/asap-l2, relocate CSE, PromQL retarget, capability_for delegation

control_plane/Cargo.toml

@@ -24,7 +24,12 @@ thiserror   = "1"
 tracing     = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
 chrono      = { version = "0.4",  features = ["serde"] }
-promql-parser  = "0.8"
+# Private mirror of GreptimeTeam/promql-parser (Apache-2.0), matching
+# ASAPController's frontend-promql -- carries local patches (limitk /
+# limit_ratio, a batch of experimental Prometheus functions) upstream
+# crates.io 0.8 doesn't have. Pinned to a commit, not the `asap` branch,
+# for the same reproducibility reason as the asap-ir pin below.
+promql-parser  = { git = "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/ProjectASAP/promql-parser", rev = "c51beafb361af4cc95ed62ae377862c660ceb757" }
 prost          = "0.13"
 bytes          = "1"
 zstd           = "0.13"
@@ -44,7 +49,20 @@ asap_types.workspace = true
 # scratchpad/artifacts/enum-unification-plan.md) -- WindowKind needs
 # Copy/Default/Hash/Display/FromStr + snake_case serde for
 # asap_types::enums::WindowKind to replace the backend's own WindowType.
+# Also satisfies everything Phase 2 (#395) needs -- asap-l2 (Step 4),
+# asap-plan/asap-sketch + Implementation::is_satisfied_by (Steps 8-9) --
+# 7fcaf91 is a strict descendant of every rev that PR pinned along the way.
+#
+# asap-plan owns the single authoritative AggIntent -> sketch-vs-exact
+# implementation decision (`boundary::implementation_for`, Cascades
+# "implementation rule" terminology -- see that crate's doc); capability_for
+# delegates to it instead of maintaining its own parallel judgment. asap-plan
+# depends only on asap-ir (per its own crate doc's layering invariant), so
+# 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 = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
+asap-l2 = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
 asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
 asap-plan = { git = "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/ProjectASAP/ASAPController", rev = "7fcaf914d87e71407c3a6d7ccac613b867f9c11b" }
 

control_plane/src/asap_tier_analysis.rs

@@ -319,10 +319,11 @@ fn collect_agg_intents(expr: &QueryExpr, out: &mut Vec<AggIntent>) {
         QueryExpr::Scan { .. } | QueryExpr::Ref { .. } => {}
         // A-variants lifted in Batch 2 of the relational migration. They
         // carry no AggIntent themselves — recurse into their children to
-        // find Aggregates further down the tree.
+        // find Aggregates further down the tree. `Partition` no longer
+        // exists in the canonical IR — its keys fold into `Aggregate.by`
+        // at construction time (`intent_algebra::lower`).
         QueryExpr::Filter { child, .. }
         | QueryExpr::Project { child, .. }
-        | QueryExpr::Partition { child, .. }
         | QueryExpr::Distinct { child, .. }
         | QueryExpr::Sort { child, .. }
         | QueryExpr::Limit { child, .. }
@@ -342,6 +343,12 @@ fn collect_agg_intents(expr: &QueryExpr, out: &mut Vec<AggIntent>) {
             collect_agg_intents(left, out);
             collect_agg_intents(right, out);
         }
+        // The PromQL-surface superset (Scalar/EvalTime/VectorFromScalar/
+        // ScalarFromVector/Relabel/InfoJoin/Sample/TimeRange/TimeShift/
+        // WindowFunc) isn't constructed by this parser today; the
+        // single-child wrappers among them carry no `AggIntent` either
+        // way, so a no-op default is safe.
+        _ => {}
     }
 }
 
@@ -1057,19 +1064,19 @@ mod tests {
         assert!(a.unsupported.is_none(), "{a:?}");
         assert_eq!(
             a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum)
+            Capability::ExactAgg(AggregationType::Increase)
         );
     }
 
     #[test]
     fn irate_binds_to_exact_agg() {
         // `irate` shares `AggFunc::Rate` with `rate` in
-        // `query_parser::promql`; both lower to `AggIntent::Sum`.
+        // `query_parser::promql`; both lower to `AggIntent::Rate`.
         let a = analyze_promql_for_asap_tier("irate(http_requests_total[5m])");
         assert!(a.unsupported.is_none(), "{a:?}");
         assert_eq!(
             a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum)
+            Capability::ExactAgg(AggregationType::Increase)
         );
     }
 
@@ -1079,7 +1086,7 @@ mod tests {
         assert!(a.unsupported.is_none(), "{a:?}");
         assert_eq!(
             a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum)
+            Capability::ExactAgg(AggregationType::Increase)
         );
     }
 
@@ -1097,14 +1104,17 @@ mod tests {
     // ── outer_fn — rate vs plain disambiguation ──────────────────────────
     //
     // Regression coverage for the PR that retired the engine's
-    // `query_contains_rate_call` raw-PromQL re-parser. The analyzer's
-    // lowerer collapses `rate(metric[r])`, `sum_over_time(metric[r])`,
-    // `sum(metric)`, and the bare selector all onto `AggIntent::Sum` /
-    // `Capability::ExactAgg(Sum)` — so the engine can't tell from the
-    // capability alone which the user wrote. The `outer_fn` field on
-    // `ASAPTierCandidate` carries the rate-vs-plain distinction so the
-    // engine's reducer dispatch is a typed branch instead of a raw-PromQL
-    // re-parse.
+    // `query_contains_rate_call` raw-PromQL re-parser. `outer_fn` is
+    // computed independently of `required_capability` — straight off the
+    // raw PromQL function name (`set_counter_fn`), not off the lowered
+    // `AggIntent` — so the engine's reducer dispatch has the rate-vs-
+    // increase-vs-plain distinction as a typed branch instead of a
+    // raw-PromQL re-parse, regardless of whether the capability
+    // computation happens to agree or differ across cases. `rate`/
+    // `irate`/`increase` now bind to distinct-from-`sum`/`sum_over_time`
+    // capabilities (`ExactAgg(Increase)` vs `ExactAgg(Sum)` — see
+    // `rate_and_increase_share_capability_but_differ_on_outer_fn` below
+    // for the pairing that still needs `outer_fn` to disambiguate).
 
     #[test]
     fn rate_candidate_carries_outer_fn_rate() {
@@ -1139,15 +1149,15 @@ mod tests {
 
     #[test]
     fn increase_candidate_carries_outer_fn_increase() {
-        // `increase(metric[r])` shares `Capability::ExactAgg(Sum)` with
-        // `rate`/`sum_over_time`; the `outer_fn` field carries the
+        // `increase(metric[r])` shares `Capability::ExactAgg(Increase)`
+        // with `rate`/`irate`; the `outer_fn` field carries the
         // distinction so the engine sums deltas in `[t-r,t]` WITHOUT the
         // rate divisor (issue #301).
         let a = analyze_promql_for_asap_tier("increase(http_requests_total[5m])");
         assert!(a.unsupported.is_none(), "{a:?}");
         assert_eq!(
             a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum),
+            Capability::ExactAgg(AggregationType::Increase),
             "{a:?}"
         );
         assert_eq!(a.candidates[0].outer_fn, OuterFn::Increase, "{a:?}");
@@ -1191,7 +1201,7 @@ mod tests {
         assert!(a.unsupported.is_none(), "{a:?}");
         assert_eq!(
             a.candidates[0].required_capability,
-            Capability::ExactAgg(AggregationType::Sum),
+            Capability::ExactAgg(AggregationType::Increase),
             "{a:?}"
         );
         assert_eq!(a.candidates[0].outer_fn, OuterFn::Rate, "{a:?}");
@@ -1200,26 +1210,58 @@ mod tests {
     }
 
     #[test]
-    fn rate_and_sum_over_time_share_capability_but_differ_on_outer_fn() {
-        // Both collapse to `Capability::ExactAgg(Sum)`; the engine MUST
-        // disambiguate via the typed `outer_fn` field, not by string-
-        // parsing the raw PromQL. This test pins the asymmetry the
-        // engine's dispatch reads off.
+    fn rate_and_sum_over_time_differ_on_capability_and_outer_fn() {
+        // Pre-PromQL-frontend-semantic-retarget, `rate`/`irate`/
+        // `increase`/`sum_over_time` all collapsed onto `AggIntent::Sum`
+        // (`Capability::ExactAgg(Sum)`), so `outer_fn` was the *only*
+        // thing that told the engine `rate(...)` needed a rate-divisor
+        // step `sum_over_time(...)` didn't. `rate`/`increase` now bind to
+        // their own dedicated `AggIntent`s (`Capability::ExactAgg(
+        // Increase)`, distinct from `sum_over_time`'s `ExactAgg(Sum)`) —
+        // a strictly more precise classification, not a regression: the
+        // capability itself now carries part of the distinction
+        // `outer_fn` used to carry alone.
         let rate = analyze_promql_for_asap_tier("rate(http_requests_total[5m])");
         let sot = analyze_promql_for_asap_tier("sum_over_time(http_requests_total[5m])");
-        assert_eq!(
+        assert_ne!(
             rate.candidates[0].required_capability, sot.candidates[0].required_capability,
-            "rate and sum_over_time should produce the same Capability"
+            "rate and sum_over_time should now bind to distinct capabilities"
         );
-        assert_ne!(
-            rate.candidates[0].outer_fn, sot.candidates[0].outer_fn,
-            "rate and sum_over_time MUST differ on outer_fn so the engine \
-             can dispatch correctly without re-parsing the raw PromQL"
+        assert_eq!(
+            rate.candidates[0].required_capability,
+            Capability::ExactAgg(AggregationType::Increase)
+        );
+        assert_eq!(
+            sot.candidates[0].required_capability,
+            Capability::ExactAgg(AggregationType::Sum)
         );
         assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate);
         assert_eq!(sot.candidates[0].outer_fn, OuterFn::SumOverTime);
     }
 
+    #[test]
+    fn rate_and_increase_share_capability_but_differ_on_outer_fn() {
+        // The pairing that now needs `outer_fn` to disambiguate: `rate`
+        // and `increase` both bind to `Capability::ExactAgg(Increase)`
+        // (rate = increase / range, an L4/reducer-level division, not a
+        // capability difference) — the engine still can't tell from the
+        // capability alone whether to apply the rate divisor, so
+        // `outer_fn` carries that distinction.
+        let rate = analyze_promql_for_asap_tier("rate(http_requests_total[5m])");
+        let inc = analyze_promql_for_asap_tier("increase(http_requests_total[5m])");
+        assert_eq!(
+            rate.candidates[0].required_capability, inc.candidates[0].required_capability,
+            "rate and increase should share the same Capability"
+        );
+        assert_ne!(
+            rate.candidates[0].outer_fn, inc.candidates[0].outer_fn,
+            "rate and increase MUST differ on outer_fn so the engine can \
+             dispatch correctly without re-parsing the raw PromQL"
+        );
+        assert_eq!(rate.candidates[0].outer_fn, OuterFn::Rate);
+        assert_eq!(inc.candidates[0].outer_fn, OuterFn::Increase);
+    }
+
     // ── outer_agg — outer aggregation operator on function results ──────
     //
     // Regression coverage for issue #296: the asap engine was rejecting

control_plane/src/deployment_model.rs

@@ -199,13 +199,14 @@ mod tests {
         let id = DeploymentModelId::asaplifecycle();
         assert!(reg.contains(&id));
         let m = reg.lookup(&id).expect("asaplifecycle must be registered");
-        // The default rule set carries the 11 engine rules. (R6
-        // HistogramQuantileFusion was retired in Step γ5 — see
-        // `optimizer::engine` module note.)
+        // The default rule set carries the 9 engine rules. (R6
+        // HistogramQuantileFusion was retired in Step γ5; R7
+        // SubqueryDecorrelation and R11 PartitionElim were retired in
+        // the `asap_ir` merge — see `optimizer::engine` module note.)
         assert_eq!(
             m.rules.len(),
-            11,
-            "asaplifecycle should ship the 11 engine rules"
+            9,
+            "asaplifecycle should ship the 9 engine rules"
         );
         // Emitter set carries the three demo emitters.
         assert!(m.emitters.has("opamp_edge_yaml"));
@@ -223,7 +224,9 @@ mod tests {
         assert!(cats.contains(&RuleCategory::Fusion));
         assert!(cats.contains(&RuleCategory::Elim));
         assert!(cats.contains(&RuleCategory::Cse));
-        assert!(cats.contains(&RuleCategory::Decorrelate));
+        // `RuleCategory::Decorrelate` has no producer left — R7
+        // SubqueryDecorrelation was retired in the `asap_ir` merge (see
+        // `optimizer::engine` module note).
     }
 
     #[test]

control_plane/src/intent_algebra/agg_intent.rs

@@ -8,6 +8,12 @@
 //! ASAPController has no tagged releases yet) and holds only what's
 //! genuinely control_plane-specific:
 //!
+//! (Phase 2: `Column`/`DataType` are also re-exported from `asap_ir` now
+//! — `schema.rs` got the full swap, not a boundary conversion, since
+//! asap_ir's version turned out to be a purely additive, backward-
+//! compatible superset. `output_column` below no longer needs a
+//! conversion layer as a result.)
+//!
 //! - **`frequency()` / `as_frequency()`** — control_plane's standalone
 //!   point-frequency-via-CMS query (`count(*) WHERE key = k`), carried
 //!   through the shared `AggIntent` as an `Extension` rather than a
@@ -43,7 +49,6 @@
 //! in scope, not off the intent.
 
 use asap_ir::intent_algebra::agg_accuracy as asap_agg_accuracy;
-use asap_ir::intent_algebra::schema::{Column as AsapColumn, DataType as AsapDataType};
 pub use asap_ir::intent_algebra::{
     agg_is_exact, agg_is_mergeable, default_cardinality, default_quantile,
     is_frequency_heavy_hitter, ranking_measure, AggIntent, MathFunc, RankingMeasure, TimeFunc,
@@ -54,43 +59,6 @@ use crate::types_v2::AccuracyTarget;
 
 const FREQUENCY_EXT_KIND: &str = "frequency";
 
-/// `control_plane::Column`/`DataType` and `asap_ir::Column`/`DataType`
-/// are structurally identical but not the same type — merging `schema.rs`
-/// itself is Phase 2 scope (it cascades into `Schema`/`QueryExpr`, used
-/// pervasively; ~38 `Column{}` literals across this repo). Convert at
-/// this boundary instead of widening this change.
-fn to_asap_dtype(dt: &DataType) -> AsapDataType {
-    match dt {
-        DataType::Int64 => AsapDataType::Int64,
-        DataType::Float64 => AsapDataType::Float64,
-        DataType::Utf8 => AsapDataType::Utf8,
-        DataType::Bool => AsapDataType::Bool,
-        DataType::Timestamp => AsapDataType::Timestamp,
-    }
-}
-
-fn from_asap_dtype(dt: &AsapDataType) -> DataType {
-    match dt {
-        AsapDataType::Int64 => DataType::Int64,
-        AsapDataType::Float64 => DataType::Float64,
-        AsapDataType::Utf8 => DataType::Utf8,
-        AsapDataType::Bool => DataType::Bool,
-        AsapDataType::Timestamp => DataType::Timestamp,
-    }
-}
-
-fn to_asap_column(c: &Column) -> AsapColumn {
-    AsapColumn::new(c.name.clone(), to_asap_dtype(&c.dtype), c.nullable)
-}
-
-fn from_asap_column(c: AsapColumn) -> Column {
-    Column {
-        name: c.name,
-        dtype: from_asap_dtype(&c.dtype),
-        nullable: c.nullable,
-    }
-}
-
 /// Construct control_plane's point-frequency-via-CMS intent. See module
 /// docs for why this is an `Extension`, not a shared first-class variant.
 pub fn frequency(accuracy: AccuracyTarget) -> AggIntent {
@@ -190,11 +158,7 @@ pub fn archive_only(intent: &AggIntent) -> bool {
 /// `"frequency"` — that's control_plane-only knowledge).
 pub fn output_column(intent: &AggIntent, input: &Column) -> Column {
     if as_frequency(intent).is_some() {
-        return Column {
-            name: "frequency".into(),
-            dtype: DataType::Int64,
-            nullable: false,
-        };
+        return Column::new("frequency", DataType::Int64, false);
     }
-    from_asap_column(intent.output_column(&to_asap_column(input)))
+    intent.output_column(input)
 }