refactor(types)!: DataWorkload as arrival plus per-source facts; G bound per source

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

@@ -4,11 +4,11 @@
 //! `Unknown` leaf named by one of the constants below, which Stage 3 binds.
 use super::*;
 
-/// Rows an exact aggregation folds (per group): bounded by the data
-/// workload's input cardinality.
+/// Rows an exact aggregation folds (per group): bounded by the series
+/// count of the sources the query reads.
 pub(crate) const INPUT_ROW_COUNT: &str = "input_row_count";
 /// Rows one group of a SQL fold over an approximate relation folds: not
-/// bounded by the data workload's input cardinality, so plan selection
+/// bounded by the series count of the sources it reads, so plan selection
 /// leaves it unbound.
 pub(crate) const GROUP_ROW_COUNT: &str = "group_row_count";
 /// Whether every input value is non-negative.

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

@@ -130,70 +130,3 @@ pub trait AccuracyEvidenceProvider {
 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: &FieldDataType,
-        _query: Option<&SketchStatistic>,
-    ) -> PropagationStats {
-        PropagationStats {
-            input_row_count: self.data.input_cardinality.value_at(self.now_ms).copied(),
-            ..PropagationStats::default()
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use asap_types::workload::{DataWorkload, Evidence, EvidenceSource};
-    #[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),
-            },
-            ..Default::default()
-        };
-        let provider = WorkloadAccuracyEvidence {
-            data: &data,
-            now_ms: 1_500,
-        };
-        let fresh = provider.propagation_stats(
-            &CompositionOperator::ExactSum,
-            &FieldDataType::ExactAggregate(
-                asap_types::post_asap::ExactKind::Sum,
-                asap_types::post_asap::ExactParams::Sum,
-            ),
-            None,
-        );
-        assert_eq!(fresh.input_row_count, Some(42));
-
-        let stale = WorkloadAccuracyEvidence {
-            data: &data,
-            now_ms: 1_501,
-        }
-        .propagation_stats(
-            &CompositionOperator::ExactSum,
-            &FieldDataType::ExactAggregate(
-                asap_types::post_asap::ExactKind::Sum,
-                asap_types::post_asap::ExactParams::Sum,
-            ),
-            None,
-        );
-        assert_eq!(stale.input_row_count, None);
-    }
-}

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

@@ -13,10 +13,7 @@ pub mod evidence;
 pub mod reconciliation;
 
 pub use budget::per_input_target;
-pub use evidence::{
-    AccuracyEvidenceProvider, NoAccuracyEvidence, OperandDomain, PropagationStats,
-    WorkloadAccuracyEvidence,
-};
+pub use evidence::{AccuracyEvidenceProvider, NoAccuracyEvidence, OperandDomain, PropagationStats};
 
 use asap_types::ir::OperatorNode;
 use asap_types::post_asap::{

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

@@ -96,12 +96,12 @@ fn resolve_cache_profile(
                 }
                 0.0
             }
-            DataArrival::ContinuouslyIngesting => evidence.result_invalidation_ratio.ok_or(
+            DataArrival::Ingesting { .. } => evidence.result_invalidation_ratio.ok_or(
                 AnalyticalCostError::InvalidCacheEvidence(
                     "continuous ingestion requires result-cache invalidation evidence",
                 ),
             )?,
-            DataArrival::Mixed | DataArrival::Unknown => {
+            DataArrival::Mixed { .. } | DataArrival::Unknown => {
                 return Err(AnalyticalCostError::InvalidCacheEvidence(
                     "cache invalidation cannot be derived for mixed or unknown data arrival",
                 ));
@@ -3811,7 +3811,13 @@ mod tests {
 
         let profile = cache_profile(100, 100);
         assert!(matches!(
-            cache_hit_ratios(&profile, 6, DataArrival::ContinuouslyIngesting),
+            cache_hit_ratios(
+                &profile,
+                6,
+                DataArrival::Ingesting {
+                    rows_per_second: 1_000.0,
+                }
+            ),
             Err(AnalyticalCostError::InvalidCacheEvidence(_))
         ));
     }
@@ -4009,7 +4015,9 @@ mod tests {
     fn streaming_invalidation_bounds_result_hits_and_changes_physical_work() {
         let (nodes, evidence) = cache_test_scan();
         let mut scope = comparison_scope();
-        scope.data_arrival = DataArrival::ContinuouslyIngesting;
+        scope.data_arrival = DataArrival::Ingesting {
+            rows_per_second: 1_000.0,
+        };
         let mut profile = cache_profile(100, 50);
         let CacheProfile::Evidence(inputs) = &mut profile else {
             unreachable!()

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

@@ -172,7 +172,7 @@ pub mod topk_reuse;
 pub use accuracy::reconciliation::AccuracyReconciliationStrategy;
 pub use accuracy::{
     AccuracyEvidenceProvider, AccuracyModel, DefaultAccuracyModel, NoAccuracyEvidence,
-    PropagationStats, WorkloadAccuracyEvidence,
+    PropagationStats,
 };
 pub use cost_model::{
     maintenance_operation_plan_cost_rate, raw_recompute_cost_rate, read_operation_plan_cost_rate,

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

@@ -138,7 +138,7 @@ fn recognize(
         }
         return Some((population, evaluation, Rc::clone(source)));
     }
-    // A bare PromQL selector carries the declared ingestion interval as a
+    // A bare PromQL selector carries its metric's sample interval as a
     // temporal input scope. Membership must expire at that horizon; retain
     // the wrapper as the maintained input so validation can check agreement.
     let (series_source, lookback_ms) = match source.non_asap() {

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

@@ -1577,8 +1577,8 @@ fn source_distinct_guarantee(
 fn fold_stats(reduction: &Reduction, input: &Rc<OperatorNode>) -> Option<PropagationStats> {
     let sign_statistic = estimates_samples_of(input).map(|metric| samples_sign_statistic(&metric));
     let stats = match reduction {
-        // The input cardinality bounds a PromQL fold's rows (series), not
-        // the rows of a SQL group.
+        // The series count of the sources read bounds a PromQL fold's rows
+        // (series), not the rows of a SQL group.
         Reduction::Reduce(_) if reads_table(input) => Some(PropagationStats {
             row_count_statistic: Some(GROUP_ROW_COUNT),
             ..PropagationStats::default()

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

@@ -71,7 +71,7 @@ impl ComparisonScope {
     pub fn validate(&self) -> Result<u64, AnalyticalCostError> {
         if !matches!(
             self.data_arrival,
-            DataArrival::AtRest | DataArrival::ContinuouslyIngesting
+            DataArrival::AtRest | DataArrival::Ingesting { .. }
         ) {
             return Err(AnalyticalCostError::UnsupportedDataArrival(
                 self.data_arrival,

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

@@ -2867,8 +2867,8 @@ mod tests {
     ) {
         use crate::physical_operator_statistics::ScalarInputStatistics;
         use asap_types::workload::{
-            BatchEntry, DataWorkload, Evidence, PlanningWorkload, Predictability, Query,
-            QueryLanguage, QueryRequirements, QueryWorkload,
+            BatchEntry, DataWorkload, PlanningWorkload, Predictability, Query, QueryLanguage,
+            QueryRequirements, QueryWorkload,
         };
         use std::cell::RefCell;
 
@@ -2887,15 +2887,12 @@ mod tests {
                 }]),
                 repeating_queries: None,
             },
-            data_workload: Some(DataWorkload {
-                data_ingestion_interval: Evidence {
-                    value: Some(DurationMs(1_000)),
-                    ..Default::default()
-                },
-                ..Default::default()
-            }),
+            data_workload: Some(DataWorkload::sampled_every(
+                asap_frontend_promql::promql_metric_names(query).unwrap_or_default(),
+                DurationMs(1_000),
+            )),
         };
-        let root = asap_frontend_promql::lower_promql_workload(&workload, 0)
+        let root = asap_frontend_promql::lower_promql_workload(&workload)
             .unwrap()
             .remove(0);
         // Read one node through every reference (as a planner that shares

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

@@ -79,8 +79,8 @@
 //!   for a whole workload). A one-shot ([`asap_types::workload::BatchEntry`])
 //!   consumer contributes to [`RecurrenceProfile::one_shot_consumers`]
 //!   instead, never to this rate.
-//! - [`UpdateRate`]: read from workload-level
-//!   [`asap_types::workload::DataWorkload::ingestion_rate`] via
+//! - [`UpdateRate`]: read from the workload-level arrival's
+//!   [`asap_types::workload::DataArrival::rows_per_second`] via
 //!   [`update_rate_from_data_workload`]. Missing evidence remains unknown;
 //!   data at rest is not assigned a fabricated update rate.
 //! - `maintenance_cost_per_update` / `summary_read_cost` /
@@ -112,7 +112,7 @@ use crate::cost_model::{Cost, CostModel, CseCandidate, ShareDecision};
 
 /// How often the *raw* data underlying a maintained summary changes —
 /// ingest/update events per second (Hz). See the module docs' provenance
-/// table: normally read from [`DataWorkload::ingestion_rate`] via
+/// table: normally read from the arrival's `rows_per_second` via
 /// [`update_rate_from_data_workload`].
 #[derive(Debug, Clone, Copy, PartialEq, PartialOrd)]
 pub struct UpdateRate(pub f64);
@@ -295,9 +295,9 @@ pub fn update_rate_from_data_workload(
     workload: &DataWorkload,
 ) -> Result<Option<UpdateRate>, RecurrenceError> {
     workload
-        .ingestion_rate
-        .value
-        .map(|rate| validate_update_rate(UpdateRate(rate.0)))
+        .arrival
+        .rows_per_second()
+        .map(|rate| validate_update_rate(UpdateRate(rate)))
         .transpose()
 }
 
@@ -323,7 +323,7 @@ pub struct RecurrenceProfile {
     pub one_shot_consumers: usize,
     /// The ingest/update rate of the raw data this target (if maintained)
     /// would be kept up to date against. `None` when no
-    /// [`DataWorkload::ingestion_rate`] evidence was available.
+    /// arrival rate (`DataArrival::rows_per_second`) was declared.
     pub update_rate: Option<UpdateRate>,
 }
 
@@ -674,11 +674,10 @@ mod tests {
     // ── update_rate_from_data_workload ───────────────────────────────────
 
     #[test]
-    fn update_rate_from_data_workload_reads_ingestion_rate_evidence() {
+    fn update_rate_from_data_workload_reads_the_arrival_rate() {
         let workload = DataWorkload {
-            ingestion_rate: asap_types::workload::Evidence {
-                value: Some(asap_types::workload::Rate(100.0)),
-                ..Default::default()
+            arrival: asap_types::workload::DataArrival::Ingesting {
+                rows_per_second: 100.0,
             },
             ..Default::default()
         };
@@ -689,9 +688,8 @@ mod tests {
     #[test]
     fn update_rate_from_data_workload_rejects_a_negative_rate() {
         let workload = DataWorkload {
-            ingestion_rate: asap_types::workload::Evidence {
-                value: Some(asap_types::workload::Rate(-0.1)),
-                ..Default::default()
+            arrival: asap_types::workload::DataArrival::Mixed {
+                rows_per_second: -0.1,
             },
             ..Default::default()
         };

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

@@ -4455,8 +4455,8 @@ impl<Id> CandidateLogicalASAPDAGs<Id> {
             recurrences.push(recurrence);
         }
         let update_rate = data_workload
-            .and_then(|data| data.ingestion_rate.value_at(now_ms))
-            .map(|rate| UpdateRate(rate.0));
+            .and_then(|data| data.arrival.rows_per_second())
+            .map(UpdateRate);
         self.recurrence_profiles(&recurrences, update_rate)
     }
 }

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

@@ -363,7 +363,7 @@ fn ingestion_eligibility(
     use IngestionPruneReason as Why;
     if !matches!(
         data.arrival,
-        DataArrival::ContinuouslyIngesting | DataArrival::Mixed
+        DataArrival::Ingesting { .. } | DataArrival::Mixed { .. }
     ) {
         return Err(Why::NotIngesting);
     }
@@ -396,7 +396,7 @@ fn ingestion_eligibility(
         // Assumption (module doc): continuously ingesting PromQL data
         // arrives in time order per series, as Prometheus assumes.
         let ordered =
-            data.arrival == DataArrival::ContinuouslyIngesting && reads_only_time_series(node);
+            matches!(data.arrival, DataArrival::Ingesting { .. }) && reads_only_time_series(node);
         if !ordered {
             return Err(Why::CounterNeedsTimeOrder);
         }
@@ -981,7 +981,9 @@ mod tests {
 
     fn arriving() -> DataWorkload {
         DataWorkload {
-            arrival: DataArrival::ContinuouslyIngesting,
+            arrival: DataArrival::Ingesting {
+                rows_per_second: 1_000.0,
+            },
             ..Default::default()
         }
     }
@@ -1109,7 +1111,9 @@ mod tests {
         };
         let table = QueryRoot::Operator(to_table(node));
         let mixed = DataWorkload {
-            arrival: DataArrival::Mixed,
+            arrival: DataArrival::Mixed {
+                rows_per_second: 1_000.0,
+            },
             ..Default::default()
         };
         for (root, data) in [(table, arriving()), (root.clone(), mixed)] {