feat(workload): normalize query demand and data evidence

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

@@ -88,6 +88,10 @@ pub struct PropagationStats {
     /// of groups a `sum` folds), for `ExactSum`/`ExactExtremum`'s union
     /// bound over per-input failures.
     pub input_row_count: Option<u64>,
+    /// Fresh key-frequency distribution evidence from the data workload.
+    /// Built-in rules preserve it for deployment-specific accuracy models;
+    /// they do not assume a favorable distribution when it is absent.
+    pub data_distribution: Option<asap_types::workload::DataDistribution>,
     /// Lower confidence bound of the kth selected TopK item, after widening
     /// the interval by the sketch's own estimation error.
     pub topk_selected_lower_bound: Option<f64>,
@@ -120,6 +124,29 @@ pub struct NoAccuracyEvidence;
 
 impl AccuracyEvidenceProvider for NoAccuracyEvidence {}
 
+/// Accuracy evidence backed by the normalized data workload. Freshness is
+/// checked at the planning time before values reach any accuracy rule.
+#[derive(Debug, Clone, Copy)]
+pub struct WorkloadAccuracyEvidence<'a> {
+    pub data: &'a asap_types::workload::DataWorkload,
+    pub now_ms: u64,
+}
+
+impl AccuracyEvidenceProvider for WorkloadAccuracyEvidence<'_> {
+    fn propagation_stats(
+        &self,
+        _op: &CompositionOperator,
+        _family: &SummaryFamilyType,
+        _query: Option<&SketchQuery>,
+    ) -> PropagationStats {
+        PropagationStats {
+            input_row_count: self.data.input_cardinality.value_at(self.now_ms).copied(),
+            data_distribution: self.data.distribution.value_at(self.now_ms).cloned(),
+            ..PropagationStats::default()
+        }
+    }
+}
+
 /// The deployment-extensible accuracy algebra. `asap-aware-mapping` ships
 /// [`DefaultAccuracyModel`]; a deployment with a proof for a composition the
 /// default rejects (a registered cross-metric conversion, say) implements
@@ -839,6 +866,7 @@ impl AccuracyBudgetAllocator for EqualSplitAllocator {
 mod tests {
     use super::*;
     use asap_types::post_asap::{GroupingStrategy, SketchKind};
+    use asap_types::workload::{DataDistribution, DataWorkload, Evidence, EvidenceSource};
 
     fn abs(bound: f64, delta: f64) -> ResultGuarantee {
         ResultGuarantee {
@@ -865,6 +893,54 @@ mod tests {
         }
     }
 
+    #[test]
+    fn workload_accuracy_evidence_uses_only_fresh_data_characteristics() {
+        let data = DataWorkload {
+            input_cardinality: Evidence {
+                value: Some(42),
+                source: EvidenceSource::Observed,
+                observed_at_ms: Some(1_000),
+                valid_for_ms: Some(500),
+            },
+            distribution: Evidence {
+                value: Some(DataDistribution::Bursty),
+                source: EvidenceSource::Observed,
+                observed_at_ms: Some(1_000),
+                valid_for_ms: Some(500),
+            },
+            ..Default::default()
+        };
+        let provider = WorkloadAccuracyEvidence {
+            data: &data,
+            now_ms: 1_500,
+        };
+        let fresh = provider.propagation_stats(
+            &CompositionOperator::ExactSum,
+            &SummaryFamilyType::ExactAggregate(
+                asap_types::post_asap::ExactKind::Sum,
+                asap_types::post_asap::ExactParams::Sum,
+            ),
+            None,
+        );
+        assert_eq!(fresh.input_row_count, Some(42));
+        assert_eq!(fresh.data_distribution, Some(DataDistribution::Bursty));
+
+        let stale = WorkloadAccuracyEvidence {
+            data: &data,
+            now_ms: 1_501,
+        }
+        .propagation_stats(
+            &CompositionOperator::ExactSum,
+            &SummaryFamilyType::ExactAggregate(
+                asap_types::post_asap::ExactKind::Sum,
+                asap_types::post_asap::ExactParams::Sum,
+            ),
+            None,
+        );
+        assert_eq!(stale.input_row_count, None);
+        assert_eq!(stale.data_distribution, None);
+    }
+
     #[test]
     fn exact_child_contributes_zero_error() {
         let local = abs(0.05, 0.01);

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

@@ -73,7 +73,7 @@
 //!
 //! ## Provenance of each new cost input
 //!
-//! - [`EvaluationRate`]: derived from [`asap_types::workload::RepeatingEntry::interval`]
+//! - [`EvaluationRate`]: derived from [`asap_types::workload::RepeatingEntry::demand`]
 //!   values of every repeating consumer reaching a target (via
 //!   [`evaluation_rate_of`], or [`crate::replacement::PlanSpace::recurrence_profiles`]
 //!   for a whole workload). A one-shot ([`asap_types::workload::BatchEntry`])
@@ -200,6 +200,12 @@ pub enum RecurrenceError {
          well-defined maintained_cost_rate"
     )]
     InvalidUpdateRate(UpdateRate),
+    #[error("invalid EvaluationRate({0:?}Hz): an evaluation rate must be finite and >= 0")]
+    InvalidEvaluationRate(EvaluationRate),
+    #[error(transparent)]
+    InvalidWorkload(#[from] asap_types::workload::WorkloadError),
+    #[error("workload entry index {index} is out of bounds for {entry_count} entries")]
+    InvalidWorkloadEntry { index: usize, entry_count: usize },
     /// A [`Horizon`] that isn't finite and strictly positive (NaN,
     /// infinite, zero, or negative) was supplied — a non-positive or
     /// infinite horizon would silently drop or invert the recurring
@@ -373,17 +379,17 @@ impl RecurrenceProfile {
 /// vs. `repeating_queries` (an interval each) split, but at the
 /// already-opaque `Id` granularity `search_workload`'s callers already use
 /// — this crate needs no more of a caller's own query identity than "which
-/// of these two recurrence kinds is this root".
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+/// of these two recurrence kinds is this root". One-time roots always use a
+/// count, so there is no second spelling for the common count-of-one case.
+#[derive(Debug, Clone, Copy, PartialEq)]
 pub enum RootRecurrence {
-    /// A one-shot (batch) root — contributes to a reached target's
-    /// [`RecurrenceProfile::one_shot_consumers`], never to its
-    /// `evaluation_rate`.
-    OneShot,
-    /// A repeating root firing every `RepetitionInterval` — contributes to
-    /// a reached target's `evaluation_rate` (`1 / interval`, aggregated via
-    /// [`evaluation_rate_of`]).
-    Repeating(RepetitionInterval),
+    /// A declared number of one-time invocations for this root.
+    OneShotCount(usize),
+    /// A repeated root normalized to evaluations per second.
+    Repeating(EvaluationRate),
+    /// No reliable recurrence evidence was supplied. It contributes no read
+    /// count or evaluation rate, but remains distinct from zero demand.
+    Unknown,
 }
 
 // ── Explanation ──────────────────────────────────────────────────────────
@@ -632,6 +638,10 @@ mod tests {
         RepetitionInterval(ms)
     }
 
+    fn repeating(ms: u32) -> RootRecurrence {
+        RootRecurrence::Repeating(evaluation_rate_of([interval(ms)]).unwrap().unwrap())
+    }
+
     // ── evaluation_rate_of ──────────────────────────────────────────────
 
     #[test]
@@ -1203,9 +1213,9 @@ mod tests {
         assert_eq!(shared_group.consumer_count, 3, "shared by all 3 roots");
 
         let root_recurrence = vec![
-            RootRecurrence::Repeating(interval(1_000)),  // 1 Hz
-            RootRecurrence::Repeating(interval(10_000)), // 0.1 Hz
-            RootRecurrence::OneShot,
+            repeating(1_000),  // 1 Hz
+            repeating(10_000), // 0.1 Hz
+            RootRecurrence::OneShotCount(1),
         ];
         let profiles = space
             .recurrence_profiles(&root_recurrence, Some(UpdateRate(5.0)))
@@ -1254,22 +1264,10 @@ mod tests {
         let update_rate = Some(UpdateRate(10.0));
 
         let frequent = space
-            .recurrence_profiles(
-                &[
-                    RootRecurrence::Repeating(interval(10)),
-                    RootRecurrence::Repeating(interval(10)),
-                ],
-                update_rate,
-            )
+            .recurrence_profiles(&[repeating(10), repeating(10)], update_rate)
             .unwrap();
         let infrequent = space
-            .recurrence_profiles(
-                &[
-                    RootRecurrence::Repeating(interval(100_000)),
-                    RootRecurrence::Repeating(interval(100_000)),
-                ],
-                update_rate,
-            )
+            .recurrence_profiles(&[repeating(100_000), repeating(100_000)], update_rate)
             .unwrap();
 
         let frequent_ranked = space
@@ -1317,14 +1315,14 @@ mod tests {
     }
 
     #[test]
-    fn recurrence_profiles_propagates_an_invalid_interval_error() {
+    fn recurrence_profiles_rejects_an_invalid_evaluation_rate() {
         let root = Rc::new(scan());
         let roots: Vec<(&str, Rc<QueryExpr>)> = vec![("only", root)];
         let space = search_workload(roots);
         let err = space
-            .recurrence_profiles(&[RootRecurrence::Repeating(interval(0))], None)
+            .recurrence_profiles(&[RootRecurrence::Repeating(EvaluationRate(f64::NAN))], None)
             .unwrap_err();
-        assert_eq!(err, RecurrenceError::InvalidInterval(interval(0)));
+        assert!(matches!(err, RecurrenceError::InvalidEvaluationRate(_)));
     }
 
     /// Issue #287 review bug 6: a length mismatch is a recoverable
@@ -1351,7 +1349,10 @@ mod tests {
         let roots: Vec<(&str, Rc<QueryExpr>)> = vec![("only", root)];
         let space = search_workload(roots);
         let err = space
-            .recurrence_profiles(&[RootRecurrence::OneShot], Some(UpdateRate(f64::NAN)))
+            .recurrence_profiles(
+                &[RootRecurrence::OneShotCount(1)],
+                Some(UpdateRate(f64::NAN)),
+            )
             .unwrap_err();
         assert!(matches!(err, RecurrenceError::InvalidUpdateRate(_)));
     }
@@ -1397,7 +1398,7 @@ mod tests {
                  target, unreachable from the original avg root's own structural children",
             );
 
-        let root_recurrence = vec![RootRecurrence::Repeating(interval(1_000))];
+        let root_recurrence = vec![repeating(1_000)];
         let profiles = space
             .recurrence_profiles(&root_recurrence, Some(UpdateRate(5.0)))
             .unwrap();
@@ -1446,7 +1447,7 @@ mod tests {
             "fixture sanity: referenced twice from the same BinaryOp parent"
         );
 
-        let root_recurrence = vec![RootRecurrence::Repeating(interval(1_000))]; // 1 Hz
+        let root_recurrence = vec![repeating(1_000)]; // 1 Hz
         let profiles = space.recurrence_profiles(&root_recurrence, None).unwrap();
         let profile = profiles.for_target(&shared_group.target);