Merge branch 'impl/sds-stack-761' into impl/sds-stack-766

Author: zzylol

Cargo.lock

@@ -14,10 +14,10 @@ version = "0.1.0"
 [workspace.dependencies]
 # Keep Planner frontends, selection, and IR on the same immutable revision.
 # Alias upstream asap-types because this workspace also defines asap_types.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
-asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
-asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
+planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
+asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
+asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
+asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
 
 # Shared external deps (used by 2+ crates)
 serde = { version = "1.0", features = ["derive"] }
@@ -37,8 +37,8 @@ arc-swap = "1.7"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 
 # Internal crates
-asap-physical-operators = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
-asap_sketch_codec = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "c899084019580eac55e6502eb641d42e748c6ff2" }
+asap-physical-operators = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
+asap_sketch_codec = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "574bef34c3792c9ebc76773eeff159ffd14194fd" }
 asap_types = { path = "crates/asap_types" }
 asap_otel_proto = { path = "crates/asap_otel_proto" }
 indexmap = { version = "2.0", features = ["serde"] }

Cargo.toml

@@ -984,6 +984,11 @@ impl BackendLocalPlanningInput {
                 &strategy,
                 &asap_aware_mapping::TargetSubDAG::new(&typed),
             );
+            proposed.candidates.extend(
+                strategy
+                    .fixed_window_rate_topk_candidates(&typed)
+                    .candidates,
+            );
             proposed.candidates.extend(direct.candidates);
             proposed.rejected.extend(direct.rejected);
             for candidate in proposed.candidates {
@@ -992,6 +997,16 @@ impl BackendLocalPlanningInput {
                 };
                 let compiled = asap_physical_operators::physical_planner::promql_rows::compile_current_series_readout(&root)
                     .or_else(|_| asap_physical_operators::physical_planner::promql_rows::compile_rate_ranking(&root).map(|(_, program)| program));
+                if let Ok(physical) = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_ranking(&root) {
+                    planner_selection_trace.push(serde_json::json!({
+                        "stage":"planner.physical_candidate", "query_id":query.query_id,
+                        "logical_root_id":crate::planner_selection::explained_root_id(&root, &query.accuracy_target),
+                        "rationale":candidate.rationale, "physical_candidate":serde_json::from_slice::<Value>(&physical.encode().map_err(|e| CompileError::Snapshot(e.to_string()))?).map_err(|e| CompileError::Snapshot(e.to_string()))?,
+                        "guarantee":root.guarantee,
+                    }));
+                    candidate_roots.push(vec![(index, root)]);
+                    continue;
+                }
                 match compiled {
                     Ok(program) => {
                         planner_selection_trace.push(serde_json::json!({
@@ -1533,7 +1548,7 @@ impl DeploymentPlanCompiler {
                 for state in &selected {
                     if matches!(&state.node.expr, SummaryExpr::SummaryAgg {
                         reduction: planner_types::pre_asap::Reduction::Reduce(keys), ..
-                    } if keys.is_empty())
+                    } if keys.is_empty()) || asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_ranking(&query.selected_plan_root).is_ok()
                     {
                         if let Some(sources) = immutable_materialization_sources(&state.node) {
                             canonical_nodes.insert(Rc::as_ptr(&state.node) as usize);
@@ -2115,7 +2130,31 @@ impl DeploymentPlanCompiler {
                     &query.selected_plan_root,
                 )
                 .ok();
+            let native_rate = native_rate.or_else(|| {
+                let physical = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_ranking(&query.selected_plan_root).ok()?;
+                fn heap(node: &Rc<SummaryNode>) -> Option<Rc<SummaryNode>> {
+                    match &node.expr {
+                        SummaryExpr::SummaryAgg { family: SummaryFamilyType::Sketch(..), .. } => Some(node.clone()),
+                        SummaryExpr::ValueOperation { child, .. } => heap(child),
+                        SummaryExpr::SummaryEstimate { summary_input, .. } => heap(summary_input),
+                        _ => None,
+                    }
+                }
+                Some((heap(&query.selected_plan_root)?, physical.query))
+            });
             let mut entry = if let Some((source, program)) = native_rate {
+                let native_state_binding = if let SummaryExpr::SummaryAgg {
+                    family: SummaryFamilyType::Sketch(..),
+                    ..
+                } = &source.expr
+                {
+                    let SummaryExpr::SummaryAgg { family, .. } = &source.expr else {
+                        unreachable!()
+                    };
+                    Some(binding(&source, family)?)
+                } else {
+                    None
+                };
                 let mut entry = crate::query_plan::compile_bound_composable_mapped(
                     query.query_id.clone(),
                     canonical.clone(),
@@ -2132,6 +2171,12 @@ impl DeploymentPlanCompiler {
                         }
                     },
                 )?;
+                if let Some(binding) = native_state_binding {
+                    entry.nodes.insert(
+                        entry.root,
+                        crate::query_plan::QueryPlanNode::ReadMaterialization { binding },
+                    );
+                }
                 let root = crate::query_plan::QueryNodeId(
                     entry.nodes.keys().map(|id| id.0).max().unwrap_or(0) + 1,
                 );
@@ -2297,7 +2342,7 @@ impl DeploymentPlanCompiler {
                 .find(|entry| entry.query_id == query_id)
                 .expect("compiled query entry exists")
                 .root;
-            let installed = super::executable_binding::install_selected_dag(
+            let mut installed = super::executable_binding::install_selected_dag(
                 query_id.clone(),
                 &compiled.dag,
                 query_plan_sink,
@@ -2313,11 +2358,35 @@ impl DeploymentPlanCompiler {
                 query_id: query_id.clone(),
                 reason,
             })?;
+            if let Ok(physical) = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_ranking(&request.queries[query_index].selected_plan_root) {
+                let program = physical.precompute.ok_or_else(|| CompileError::Snapshot("fixed-window candidate has no maintenance program".into()))?;
+                let sink = PostAsapNodeId(u32::try_from(program.roots()[0]).map_err(|_| CompileError::Snapshot("maintenance sink overflow".into()))?);
+                installed.native_programs.insert(sink, serde_json::from_slice(&program.encode().map_err(|e| CompileError::Snapshot(e.to_string()))?).map_err(|e| CompileError::Snapshot(e.to_string()))?);
+            }
             query_plan
                 .selected_dags
                 .insert(query_id.clone(), installed.document.clone());
             installed_dags.insert(query_id, installed);
         }
+        let maintenance_lookbacks = materializations
+            .iter()
+            .filter_map(|target| {
+                target
+                    .derived_input
+                    .as_ref()
+                    .map(|input| (input, target.stored_window_ms()))
+            })
+            .flat_map(|(input, window)| input.inputs.iter().map(move |source| (*source, window)))
+            .fold(
+                BTreeMap::<asap_types::sds::StoredOutputId, u64>::new(),
+                |mut windows, (source, window)| {
+                    windows
+                        .entry(source)
+                        .and_modify(|old| *old = (*old).max(window))
+                        .or_insert(window);
+                    windows
+                },
+            );
         for materialization in &mut materializations {
             let fingerprint = materialization.policy_fingerprint();
             let max_lookback_ms = query_plan
@@ -2326,6 +2395,7 @@ impl DeploymentPlanCompiler {
                 .flat_map(QueryPlanEntry::materialization_bindings)
                 .filter(|binding| binding.materialization.fingerprint() == fingerprint)
                 .filter_map(|binding| binding.readout_lookback_ms)
+                .chain(maintenance_lookbacks.get(&fingerprint.into()).copied())
                 .max();
             if let Some(lookback_ms) = max_lookback_ms {
                 materialization.num_aggregates_to_retain = Some(retained_state_count(
@@ -3434,6 +3504,11 @@ pub(crate) fn retained_partition_count(
             A::Increase | A::Rate | A::Min | A::Max
         )
         || !materialization.grouping_labels.names().is_empty()
+        || (materialization.derived_input.is_some()
+            && matches!(
+                materialization.aggregation_type,
+                A::CountMinSketchWithHeap | A::CountSketchWithHeap
+            ))
     {
         u128::from(input_cardinality.unwrap_or(1).max(1))
     } else {
@@ -3761,6 +3836,21 @@ fn immutable_materialization_sources(node: &SummaryNode) -> Option<Vec<Rc<Summar
         // because Planner made the exact accumulator boundary explicit.
         return None;
     };
+    if matches!(&node.expr, SummaryExpr::SummaryAgg { family: SummaryFamilyType::Sketch(kind, _), .. }
+        if matches!(kind.algorithm(), SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap))
+    {
+        if let SummaryExpr::ValueOperation {
+            child: source,
+            operation: planner_types::post_asap::ValueOperation::FinalizeExactAccumulator,
+            timing: ExecutionTiming::IngestionTime,
+        } = &child.expr
+        {
+            if matches!(&source.expr, SummaryExpr::SummaryAgg { family: SummaryFamilyType::ExactAggregate(ExactKind::Rate, _), reduction: planner_types::pre_asap::Reduction::PerEntity, child, .. } if matches!(child.expr, SummaryExpr::KeepPreAsap(_)))
+            {
+                return Some(vec![source.clone()]);
+            }
+        }
+    }
     if input.item.is_some()
         || !matches!(
             input.weight,
@@ -3934,6 +4024,22 @@ fn scoped_materialization(
     let SummaryExpr::SummaryAgg { reduction, .. } = &node.expr else {
         anyhow::bail!("materialization lacks SummaryAgg partition contract");
     };
+    if immutable_materialization_sources(node).is_some_and(|sources| {
+        sources.iter().any(|source| {
+            matches!(
+                &source.expr,
+                SummaryExpr::SummaryAgg {
+                    family: SummaryFamilyType::ExactAggregate(
+                        planner_types::post_asap::ExactKind::Rate,
+                        _
+                    ),
+                    ..
+                }
+            )
+        })
+    }) {
+        config.grouping_labels = asap_types::KeyByLabelNames { labels: Vec::new() }.into();
+    }
     config.partitioning = Some(match reduction {
         planner_types::pre_asap::Reduction::PerEntity => {
             asap_types::sds::PopulationPartitioning::PerEntity

control_plane/src/physical/compiler.rs

@@ -55,6 +55,7 @@ pub fn install_selected_dag(
     }
     precompute_sinks.sort();
     let installed = InstalledPostAsapDag {
+        native_programs: std::collections::BTreeMap::new(),
         document: OwnedPostAsapDag::from_executable(query_id, dag)?,
         binding: BackendExecutableBinding {
             nodes,

control_plane/src/physical/executable_binding.rs

@@ -66,6 +66,14 @@ pub fn render(plan: &CompiledPhysicalPlan) -> String {
                 edge.role
             ));
         }
+        for (sink, program) in &installed.native_programs {
+            render_native(
+                &mut dot,
+                &format!("maintenance_{dag_index}_{}", sink.0),
+                "Planner maintenance operators",
+                program,
+            );
+        }
         dot.push_str("    }\n");
     }
     dot.push_str("  }\n");
@@ -104,12 +112,48 @@ pub fn render(plan: &CompiledPhysicalPlan) -> String {
                 ));
             }
         }
+        if let Some(program) = &entry.physical_dag {
+            render_native(
+                &mut dot,
+                &format!("native_query_{query_index}"),
+                "Planner query operators",
+                program,
+            );
+        }
         dot.push_str("  }\n");
     }
     dot.push_str("}\n");
     dot
 }
 
+fn render_native(dot: &mut String, prefix: &str, label: &str, program: &serde_json::Value) {
+    let Some(nodes) = program["nodes"].as_object() else {
+        return;
+    };
+    dot.push_str(&format!(
+        "    subgraph cluster_{prefix} {{\n      label=\"{}\";\n",
+        escape(label)
+    ));
+    for (id, node) in nodes {
+        let operator = node.get("Operator");
+        let label = operator
+            .map(|op| op["operator"]["kind"].to_string())
+            .unwrap_or_else(|| "Bound physical input".into());
+        emit_node(
+            dot,
+            &format!("{prefix}_{id}"),
+            &format!("#{id}\n{label}"),
+            "",
+        );
+        if let Some(inputs) = operator.and_then(|op| op["inputs"].as_array()) {
+            for input in inputs {
+                dot.push_str(&format!("      {prefix}_{input} -> {prefix}_{id};\n"));
+            }
+        }
+    }
+    dot.push_str("    }\n");
+}
+
 fn materialization_node(id: u64) -> String {
     format!("materialization_{id:016x}")
 }

control_plane/src/physical/plan_dot.rs

@@ -215,6 +215,28 @@ pub fn manifest(
             }
         }
     }
+    for installed in plan.precompute_plan.executable_dags.values() {
+        for (sink, program) in &installed.native_programs {
+            let Some(asap_types::executable_plan::BackendNodeBinding::Materialization {
+                stored_output,
+            }) = installed.binding.node(*sink)
+            else {
+                return Err(invalid("native maintenance sink is unbound"));
+            };
+            let config = plan
+                .precompute_plan
+                .materializations
+                .iter()
+                .find(|m| m.policy_fingerprint() == stored_output.fingerprint())
+                .ok_or_else(|| invalid("native maintenance config is absent"))?;
+            add(
+                format!("maintenance:{}", stored_output.0),
+                json!({"physical_program":program,"stored_output":stored_output,"window_ms":config.stored_window_ms(),"interval_ms":config.slide_interval.saturating_mul(1000)}),
+                "horizon",
+                1.0,
+            );
+        }
+    }
     for rule in &plan.transmission_plan.rules {
         add(
             format!("transport:{}:{}", rule.producer_id, rule.materialization.0),