fix(promql): lower workloads with ingestion intervals

Author: milindsrivastava1997

crates/frontend-promql/src/error.rs

@@ -1,6 +1,7 @@
 use std::fmt;
 
 use asap_types::pre_asap::ResolveTreeError;
+use asap_types::workload::WorkloadError;
 
 /// Errors from lowering a PromQL query (parse → the canonical, unresolved
 /// tree, built directly →
@@ -13,6 +14,8 @@ use asap_types::pre_asap::ResolveTreeError;
 /// shared, so neither front end pulls the other's parser.
 #[derive(Debug)]
 pub enum PromqlError {
+    /// The workload omitted information required for plan-ready PromQL lowering.
+    InvalidWorkload(WorkloadError),
     /// The `promql-parser` crate rejected the query string (parse failure).
     Parse(String),
     /// A PromQL function (`rate`, `*_over_time`, …) not supported in this version.
@@ -36,6 +39,7 @@ pub enum PromqlError {
 impl fmt::Display for PromqlError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
+            Self::InvalidWorkload(e) => write!(f, "invalid PromQL workload: {e}"),
             Self::Parse(e) => write!(f, "PromQL parse error: {e}"),
             Self::UnsupportedFunction(n) => write!(f, "unsupported PromQL function: {n}"),
             Self::UnsupportedAggregateOp(n) => write!(f, "unsupported PromQL aggregate op: {n}"),
@@ -56,6 +60,12 @@ impl From<ResolveTreeError> for PromqlError {
     }
 }
 
+impl From<WorkloadError> for PromqlError {
+    fn from(e: WorkloadError) -> Self {
+        Self::InvalidWorkload(e)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

crates/frontend-promql/src/lib.rs

@@ -14,65 +14,111 @@ pub mod promql;
 
 use asap_types::pre_asap::resolve_root;
 use asap_types::pre_asap::QueryExpr;
-use asap_types::types::AccuracyTarget;
-use asap_types::workload::{QueryLanguage, QueryWorkload};
+use asap_types::workload::{DurationMs, QueryLanguage, QueryWorkload};
 
 pub use error::PromqlError;
 pub use histogram::{HistogramCatalog, HistogramKind};
 pub use promql::PromqlLowerer;
 
-/// Lower a single PromQL query string to the canonical, resolved `QueryExpr`.
+/// Lower every normalized PromQL workload entry to a plan-ready `QueryExpr`.
 ///
-/// `accuracy` is threaded onto every approximate intent (`Count`, `Quantile`,
-/// `Cardinality`, `TopK`). The returned tree carries a self-contained `Schema`
-/// on its `Scan`; call [`QueryExpr::output_schema`] for any node's schema.
-///
-/// `histogram_quantile` discrimination uses the structural heuristic; to drive
-/// it from declared sample types instead, use [`lower_promql_with_histograms`].
-pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Result<QueryExpr, PromqlError> {
-    let unresolved = PromqlLowerer::lower(query, &accuracy)?;
-    let resolved = resolve_root(&unresolved)?;
-    Ok(resolved)
+/// PromQL workloads must declare a non-zero `data_ingestion_interval`; it is
+/// injected around each bare instant selector. Explicit range selectors keep
+/// their query-specified range.
+pub fn lower_promql_workload(workload: &QueryWorkload) -> Result<Vec<QueryExpr>, PromqlError> {
+    if !matches!(workload.language, QueryLanguage::PromQL) {
+        return Err(PromqlError::WrongLanguage(format!(
+            "{:?}",
+            workload.language
+        )));
+    }
+    workload.validate()?;
+    let DurationMs(interval_ms) = workload
+        .data_workload
+        .as_ref()
+        .expect("validated PromQL workload has data_workload")
+        .data_ingestion_interval
+        .value
+        .expect("validated PromQL workload has data_ingestion_interval");
+    workload
+        .entries()
+        .map(|entry| {
+            let unresolved = PromqlLowerer::lower_with_ingestion_interval(
+                &entry.query.0,
+                &entry.requirements.accuracy.target(),
+                std::time::Duration::from_millis(interval_ms),
+            )?;
+            Ok(resolve_root(&unresolved)?)
+        })
+        .collect()
 }
 
-/// Like [`lower_promql`], but consults `histograms` to decide whether a
-/// `histogram_quantile` argument is sketch-able (generic `Quantile`) or a
-/// classic-bucket interpolation (`HistogramQuantile`) — a type-driven decision
-/// instead of the structural heuristic (issue #79). Metrics absent from the
-/// catalog still fall back to the heuristic.
-pub fn lower_promql_with_histograms(
-    query: &str,
-    accuracy: AccuracyTarget,
-    histograms: HistogramCatalog,
-) -> Result<QueryExpr, PromqlError> {
-    let _guard = histogram::CatalogGuard::install(histograms);
-    lower_promql(query, accuracy)
-}
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
 
-/// Lower every PromQL batch entry in `workload` to a `QueryExpr`.
-///
-/// One `Result` per entry — errors are per-query, not fatal for the batch.
-/// Returns an empty `Vec` if `workload.query_batch` is absent or empty, and a
-/// `WrongLanguage` error for every entry if the workload language is not PromQL.
-pub fn lower_promql_batch(workload: &QueryWorkload) -> Vec<Result<QueryExpr, PromqlError>> {
-    let entries = match &workload.query_batch {
-        Some(e) if !e.is_empty() => e,
-        _ => return vec![],
+    use asap_types::pre_asap::QueryExpr;
+    use asap_types::workload::{
+        BatchEntry, DataWorkload, Evidence, Query, QueryRequirements, TimeSelection,
     };
 
-    if !matches!(workload.language, QueryLanguage::PromQL) {
-        let lang = format!("{:?}", workload.language);
-        return entries
-            .iter()
-            .map(|_| Err(PromqlError::WrongLanguage(lang.clone())))
-            .collect();
+    use super::*;
+
+    fn workload(query: &str) -> QueryWorkload {
+        QueryWorkload {
+            language: QueryLanguage::PromQL,
+            query_batch: Some(vec![BatchEntry {
+                query: Query(query.into()),
+                requirements: QueryRequirements::default(),
+                predictability: Default::default(),
+                invocations: 1,
+                execute_at: None,
+                time_selection: TimeSelection::default(),
+            }]),
+            repeating_queries: None,
+            data_workload: Some(DataWorkload {
+                data_ingestion_interval: Evidence {
+                    value: Some(DurationMs(1_000)),
+                    ..Default::default()
+                },
+                ..Default::default()
+            }),
+        }
     }
 
-    entries
-        .iter()
-        .map(|entry| {
-            let accuracy = entry.requirements.accuracy.target();
-            lower_promql(&entry.query.0, accuracy)
-        })
-        .collect()
+    #[test]
+    fn instant_selector_uses_declared_ingestion_interval() {
+        let query = lower_promql_workload(&workload("sum by (job) (data)")).unwrap();
+        let QueryExpr::Aggregate { child, .. } = &query[0] else {
+            panic!("expected aggregate")
+        };
+        assert!(
+            matches!(child.as_ref(), QueryExpr::TimeRange { range, child }
+            if *range == Duration::from_secs(1) && matches!(child.as_ref(), QueryExpr::Scan { .. }))
+        );
+    }
+
+    #[test]
+    fn explicit_range_selector_keeps_its_query_range() {
+        let query = lower_promql_workload(&workload("sum_over_time(data[5m])")).unwrap();
+        let QueryExpr::Aggregate { child, .. } = &query[0] else {
+            panic!("expected aggregate")
+        };
+        assert!(
+            matches!(child.as_ref(), QueryExpr::TimeRange { range, child }
+            if *range == Duration::from_secs(300) && matches!(child.as_ref(), QueryExpr::Scan { .. }))
+        );
+    }
+
+    #[test]
+    fn workload_without_interval_fails_loudly() {
+        let mut workload = workload("sum(data)");
+        workload.data_workload = Some(DataWorkload::default());
+        assert!(matches!(
+            lower_promql_workload(&workload),
+            Err(PromqlError::InvalidWorkload(
+                asap_types::workload::WorkloadError::MissingDataIngestionInterval
+            ))
+        ));
+    }
 }

crates/frontend-promql/src/promql.rs

@@ -191,11 +191,24 @@ impl PromqlLowerer {
         check_depth(&ast, MAX_DEPTH)?;
         walk(&ast)
     }
+
+    pub(crate) fn lower_with_ingestion_interval(
+        query: &str,
+        accuracy: &AccuracyTarget,
+        interval: Duration,
+    ) -> Result<Unresolved> {
+        let _guard = AccuracyGuard::install(accuracy.clone());
+        let _interval = IngestionIntervalGuard::install(interval);
+        let ast = parser::parse(query).map_err(LoweringError::Parse)?;
+        check_depth(&ast, MAX_DEPTH)?;
+        walk(&ast)
+    }
 }
 
 std::thread_local! {
     static ACCURACY: std::cell::RefCell<AccuracyTarget> =
         const { std::cell::RefCell::new(AccuracyTarget::Exact) };
+    static INGESTION_INTERVAL: std::cell::RefCell<Option<Duration>> = const { std::cell::RefCell::new(None) };
 }
 
 /// RAII guard installing `accuracy` as the ambient accuracy target for the
@@ -221,6 +234,28 @@ fn current_accuracy() -> AccuracyTarget {
     ACCURACY.with(|a| a.borrow().clone())
 }
 
+struct IngestionIntervalGuard(Option<Duration>);
+
+impl IngestionIntervalGuard {
+    fn install(interval: Duration) -> Self {
+        Self(INGESTION_INTERVAL.with(|current| current.replace(Some(interval))))
+    }
+}
+
+impl Drop for IngestionIntervalGuard {
+    fn drop(&mut self) {
+        INGESTION_INTERVAL.with(|current| *current.borrow_mut() = self.0.take());
+    }
+}
+
+fn current_ingestion_interval() -> Duration {
+    INGESTION_INTERVAL.with(|current| {
+        current
+            .borrow()
+            .expect("ingestion interval is installed for workload lowering")
+    })
+}
+
 /// Bounded depth check over the parser AST: errors once nesting would exceed
 /// `budget` frames, descending into every child expression.
 fn check_depth(expr: &Expr, budget: usize) -> Result<()> {
@@ -299,7 +334,7 @@ fn walk(expr: &Expr) -> Result<Unresolved> {
         }),
         Expr::VectorSelector(vs) => {
             let (metric, matchers, shift) = vs_parts(vs)?;
-            Ok(filtered_source(metric, matchers, shift))
+            Ok(instant_source(metric, matchers, shift))
         }
         Expr::MatrixSelector(ms) => {
             let (metric, matchers, shift) = vs_parts(&ms.vs)?;
@@ -1421,7 +1456,7 @@ fn lower_inner_call(call: &Call) -> Result<Inner> {
 fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved> {
     match outer {
         Outer::None => match &inner.func {
-            None => Ok(filtered_source(inner.metric, inner.matchers, inner.shift)),
+            None => Ok(instant_source(inner.metric, inner.matchers, inner.shift)),
             Some(f) => {
                 let intent = inner_intent(f);
                 Ok(windowed_aggregate(inner, keys, intent))
@@ -1463,7 +1498,7 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved>
             // wrapped in a reducing aggregate (issue #86).
             let base = match inner.func.as_ref().map(inner_intent) {
                 Some(intent) => windowed_aggregate(inner, vec![], intent),
-                None => filtered_source(inner.metric, inner.matchers, inner.shift),
+                None => instant_source(inner.metric, inner.matchers, inner.shift),
             };
             Ok(Unresolved::PromqlSeriesSample {
                 by: keys.into(),
@@ -1526,7 +1561,7 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved>
                 // `Sort.partition_by` can rank within each group (issue #12).
                 let base = match inner.func.as_ref().map(inner_intent) {
                     Some(intent) => windowed_aggregate(inner, vec![], intent),
-                    None => filtered_source(inner.metric, inner.matchers, inner.shift),
+                    None => instant_source(inner.metric, inner.matchers, inner.shift),
                 };
                 let sorted = Unresolved::Sort {
                     keys: vec![SortKey {
@@ -1588,7 +1623,10 @@ fn windowed_aggregate(
             range: w,
             child: Rc::new(base),
         },
-        None => base,
+        None => Unresolved::TimeRange {
+            range: current_ingestion_interval(),
+            child: Rc::new(base),
+        },
     };
     let reduction = reduction_for(&keys, &intent, &child);
     Unresolved::Aggregate {
@@ -1641,6 +1679,13 @@ fn filtered_source(metric: String, matchers: Vec<Unresolved>, shift: TimeShift)
     }
 }
 
+fn instant_source(metric: String, matchers: Vec<Unresolved>, shift: TimeShift) -> Unresolved {
+    Unresolved::TimeRange {
+        range: current_ingestion_interval(),
+        child: Rc::new(filtered_source(metric, matchers, shift)),
+    }
+}
+
 /// Count vector elements regardless of their sample values.
 fn count() -> AggIntent<ColumnRef> {
     AggIntent::Count {

crates/types/src/workload.rs

@@ -530,6 +530,9 @@ pub struct Rate(pub f64);
 #[serde(deny_unknown_fields)]
 pub struct DataWorkload {
     pub arrival: DataArrival,
+    /// Cadence at which each PromQL source supplies a sample. PromQL instant
+    /// selectors use this as their explicit selection horizon.
+    pub data_ingestion_interval: Evidence<DurationMs>,
     pub ingestion_volume: Evidence<u64>,
     pub ingestion_rate: Evidence<Rate>,
     pub input_cardinality: Evidence<u64>,
@@ -586,6 +589,19 @@ impl QueryWorkload {
             }
             validate_optional_rate(data.ingestion_rate.value)?;
         }
+        if matches!(self.language, QueryLanguage::PromQL) {
+            let data = self
+                .data_workload
+                .as_ref()
+                .ok_or(WorkloadError::MissingPromqlDataWorkload)?;
+            let interval = data
+                .data_ingestion_interval
+                .value
+                .ok_or(WorkloadError::MissingDataIngestionInterval)?;
+            if interval.0 == 0 {
+                return Err(WorkloadError::ZeroDataIngestionInterval);
+            }
+        }
         Ok(())
     }
 }
@@ -638,6 +654,12 @@ fn validate_rate(rate: Rate) -> Result<Rate, WorkloadError> {
 
 #[derive(Debug, Clone, Copy, PartialEq, thiserror::Error)]
 pub enum WorkloadError {
+    #[error("a PromQL workload requires data_workload")]
+    MissingPromqlDataWorkload,
+    #[error("a PromQL workload requires data_ingestion_interval")]
+    MissingDataIngestionInterval,
+    #[error("data_ingestion_interval must be greater than zero")]
+    ZeroDataIngestionInterval,
     #[error("a one-time query must have at least one invocation")]
     ZeroInvocations,
     #[error("a fixed repetition interval must be greater than zero")]
@@ -665,7 +687,13 @@ mod tests {
             language: QueryLanguage::PromQL,
             query_batch: None,
             repeating_queries: None,
-            data_workload: None,
+            data_workload: Some(DataWorkload {
+                data_ingestion_interval: Evidence {
+                    value: Some(DurationMs(1_000)),
+                    ..Default::default()
+                },
+                ..Default::default()
+            }),
         }
     }