feat: plan maintained populations through summary lifecycles

crates/asap-aware-mapping/src/lib.rs

@@ -221,13 +221,14 @@ pub use summary_maintenance_dag_export::{
 };
 pub use summary_maintenance_lifecycle::{
     assemble_selected_dag_with_summary_maintenance_lifecycles,
-    global_selection_with_summary_maintenance_lifecycles, plan_summary_maintenance_lifecycles,
-    SummaryMaintenanceCapabilities, SummaryMaintenanceDeployment,
-    SummaryMaintenanceLifecycleAlternative, SummaryMaintenanceLifecycleAssemblyError,
-    SummaryMaintenanceLifecycleCapabilities, SummaryMaintenanceLifecycleCostInputs,
-    SummaryMaintenanceLifecyclePlan, SummaryMaintenanceLifecyclePlanError,
-    SummaryMaintenanceLifecycleRejection, SummaryMaintenanceLifecycleSelectionError,
-    WorkloadDemand,
+    enumerate_summary_maintenance_lifecycles, global_selection_with_summary_maintenance_lifecycles,
+    plan_summary_maintenance_lifecycles, SummaryMaintenanceCapabilities,
+    SummaryMaintenanceDeployment, SummaryMaintenanceLifecycleAlternative,
+    SummaryMaintenanceLifecycleAssemblyError, SummaryMaintenanceLifecycleCandidates,
+    SummaryMaintenanceLifecycleCapabilities, SummaryMaintenanceLifecycleChoiceError,
+    SummaryMaintenanceLifecycleCostInputs, SummaryMaintenanceLifecyclePlan,
+    SummaryMaintenanceLifecyclePlanError, SummaryMaintenanceLifecycleRejection,
+    SummaryMaintenanceLifecycleSelectionError, SummaryMaintenanceTimingError, WorkloadDemand,
 };
 pub use topk_reuse::TopKLimitReuseStrategy;
 

crates/asap-aware-mapping/src/maintained_population.rs

@@ -258,11 +258,15 @@ impl MaintainedPopulationStrategy {
             schema: input_schema.clone(),
             guarantee: Some(ResultGuarantee::exact("source samples")),
         });
+        // Query time is only the initial layout: whether the population is
+        // retained at ingestion or rebuilt per query is its lifecycle choice
+        // (`SummaryMaintenanceLifecyclePlan::execution_timed_dag`). The readout
+        // and projection above it are query-time by construction.
         let maintained = Rc::new(SummaryNode {
             expr: SummaryExpr::ValueOperation {
                 child: scan,
                 operation: ValueOperation::MaintainPopulation { population },
-                timing: ExecutionTiming::IngestionTime,
+                timing: ExecutionTiming::QueryTime,
             },
             schema: input_schema,
             guarantee: Some(ResultGuarantee::exact(
@@ -435,6 +439,31 @@ mod tests {
         assert_eq!(p.grouping, ["instance"]);
         assert_eq!(p.matchers[0].operation, CurrentSeriesMatch::Regex);
     }
+    // Population timing is a lifecycle choice: a retained or rebuilt
+    // population both validate, while its readout must stay at query time.
+    #[test]
+    fn population_timing_is_not_structural() {
+        let root = lower("topk(5,a)");
+        let candidate = MaintainedPopulationStrategy::new(std::slice::from_ref(&root))
+            .candidate(&root)
+            .unwrap();
+        let with_timings = |population: ExecutionTiming, readout: ExecutionTiming| {
+            let mut node = (*candidate).clone();
+            let SummaryExpr::ValueOperation { child, timing, .. } = &mut node.expr else {
+                unreachable!()
+            };
+            *timing = readout;
+            let SummaryExpr::ValueOperation { timing, .. } = &mut Rc::make_mut(child).expr else {
+                unreachable!()
+            };
+            *timing = population;
+            compile_post_asap_dag(&Rc::new(node))
+        };
+        use ExecutionTiming::{IngestionTime, QueryTime};
+        assert!(with_timings(IngestionTime, QueryTime).is_ok());
+        assert!(with_timings(QueryTime, QueryTime).is_ok());
+        assert!(with_timings(IngestionTime, IngestionTime).is_err());
+    }
     // A readout cannot reinterpret arbitrary rows as maintained state or exceed its producer's contract.
     #[test]
     fn malformed_population_dags_fail_closed() {