refactor(stage0, types): memo hash-consing, candidate storage, and aggregate coverage with timestamp intervals and chains (#1207 PR 2)

crates/executor/src/operators/series_window.rs

@@ -139,11 +139,10 @@ pub(super) fn anchor(
     match bound {
         None => Ok(literal),
         Some(bound) => {
-            let (start, end) = context
+            let (_, end) = context
                 .query_range()
-                .ok_or_else(|| invalid("@ start() and @ end() require query range bounds"))?;
+                .ok_or_else(|| invalid("@ end() requires query range bounds"))?;
             match bound {
-                AtModifier::Start => Ok(Some(start)),
                 AtModifier::End => Ok(Some(end)),
                 AtModifier::Timestamp(_) => Err(invalid("invalid range-bound anchor")),
             }

crates/executor/src/runtime/context.rs

@@ -62,7 +62,7 @@ impl RunContext {
             }),
         })
     }
-    /// Supply the outer range query bounds for PromQL `@ start()` / `@ end()`
+    /// Supply the outer range query bounds for PromQL `@ end()`
     /// and for `@`-pinned calls, which are evaluated once at the query start.
     /// A range query run step by step must always supply them; without bounds
     /// each step is treated as its own instant query. For an instant query,

crates/frontend-common/src/canonicalize.rs

@@ -1,7 +1,7 @@
 //! Post-lowering canonicalization of the operator DAG.
 //!
 //! Rewrites equivalent spellings of a query into one shape, so that common
-//! sub-DAG sharing (Stage 0 hash-consing, `cse` in `asap_stage0_memo`) can match
+//! sub-DAG sharing (hash-consing, `Memo::find_or_add_group`) can match
 //! them by structural equality, whichever front end produced them.
 //!
 //! ## Heavy-hitter promotion

crates/frontend-common/src/resolve.rs

@@ -487,11 +487,21 @@ fn resolve_expr_in(
         S::Not(e) => ScalarExpr::Not(bx(e)?),
         S::IsNull(e) => ScalarExpr::IsNull(bx(e)?),
         S::IsNotNull(e) => ScalarExpr::IsNotNull(bx(e)?),
-        S::Cast { expr, to, try_cast } => ScalarExpr::Cast {
-            expr: bx(expr)?,
-            to: to.clone(),
-            try_cast: *try_cast,
-        },
+        // A cast to the type its operand already has is the operand: SQL's
+        // `ts >= '2026-11-01'` casts the Timestamp column `ts` to DataFusion's
+        // nanosecond timestamp, which is the same IR type.
+        S::Cast { expr, to, try_cast } => {
+            let operand = bx(expr)?;
+            if matches!(operand.scalar_type(schema), Ok((dtype, _)) if dtype == *to) {
+                *operand
+            } else {
+                ScalarExpr::Cast {
+                    expr: operand,
+                    to: to.clone(),
+                    try_cast: *try_cast,
+                }
+            }
+        }
         S::InList {
             expr,
             list,

crates/frontend-promql/src/lib.rs

@@ -107,6 +107,9 @@ pub fn lower_promql_query_workload(
                 &entry.requirements.accuracy.clone(),
                 &sample_interval_of,
                 &metric_type_of,
+                entry
+                    .range_evaluation
+                    .map_or(0, |range| range.span_ms as i64),
             )?;
             Ok(root)
         })
@@ -118,11 +121,12 @@ mod tests {
     use std::rc::Rc;
     use std::time::Duration;
 
+    use asap_types::ir::operator::operator_properties::{AtModifier, TimeShift};
     use asap_types::ir::operator::Source;
-    use asap_types::ir::{NonASAPOp, OperatorNode, TimeRangeKind};
+    use asap_types::ir::{NonASAPOp, OperatorNode, QueryRoot, TimeRangeKind};
     use asap_types::workload::{
         BatchEntry, DataArrival, DataWorkload, PlanningWorkload, Query, QueryRequirements,
-        QueryWorkload, SourceFacts, WorkloadError,
+        QueryWorkload, RangeEvaluation, SourceFacts, WorkloadError,
     };
 
     use super::*;
@@ -367,4 +371,35 @@ mod tests {
         )
         .is_empty());
     }
+
+    /// `@ start()` of a range query that spans 1 h is `@ end() offset 1h`:
+    /// the range starts 1 h before it ends.
+    #[test]
+    fn at_start_is_at_end_offset_by_the_query_range() {
+        let mut workload = workload("sum_over_time(data[5m] @ start() offset 1m)");
+        if let Some(batch) = workload.query_workload.query_batch.as_mut() {
+            batch[0].range_evaluation = Some(RangeEvaluation {
+                step_ms: 60_000,
+                span_ms: 3_600_000,
+            });
+        }
+        let roots = lower_promql_query_workload(&workload).unwrap();
+        let [QueryRoot::Operator(root)] = &roots[..] else {
+            panic!("one operator root");
+        };
+        let shifts: Vec<TimeShift> = OperatorNode::reachable(root)
+            .iter()
+            .filter_map(|node| match node.non_asap()? {
+                NonASAPOp::TimeShift { shift, .. } => Some(*shift),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(
+            shifts,
+            [TimeShift {
+                offset_ms: 3_660_000,
+                at: Some(AtModifier::End),
+            }]
+        );
+    }
 }

crates/frontend-promql/src/promql.rs

@@ -97,6 +97,9 @@ pub(crate) struct PromqlLowerer<'a> {
     /// What a bare instant selector over a metric looks back; `None` when
     /// the metric declares no sample interval.
     sample_interval_of: &'a dyn Fn(&str) -> Option<Duration>,
+    /// The duration D of the query range, in ms (0 for an instant query):
+    /// `@ start()` is lowered as `@ end() offset D`.
+    query_range_ms: i64,
 }
 
 #[derive(Debug, Clone)]
@@ -141,6 +144,7 @@ impl PromqlLowerer<'_> {
         accuracy: &AccuracyTarget,
         sample_interval_of: &dyn Fn(&str) -> Option<Duration>,
         metric_type_of: &dyn Fn(&str) -> Option<PrometheusMetricType>,
+        query_range_ms: i64,
     ) -> Result<asap_types::ir::QueryRoot> {
         let ast = parser::parse(query).map_err(LoweringError::Parse)?;
         check_depth(&ast, MAX_DEPTH)?;
@@ -149,6 +153,7 @@ impl PromqlLowerer<'_> {
         let lowerer = PromqlLowerer {
             accuracy: accuracy.clone(),
             sample_interval_of,
+            query_range_ms,
         };
         if metrics
             .iter()
@@ -240,7 +245,7 @@ impl PromqlLowerer<'_> {
                     child: Rc::new(self.walk(&sq.expr)?),
                 };
                 // `offset`/`@` move the whole subquery, including its step grid.
-                let shift = time_shift(sq.offset.as_ref(), sq.at.as_ref())?;
+                let shift = time_shift(sq.offset.as_ref(), sq.at.as_ref(), self.query_range_ms)?;
                 Ok(if shift.is_identity() {
                     subquery
                 } else {
@@ -251,11 +256,11 @@ impl PromqlLowerer<'_> {
                 })
             }
             Expr::VectorSelector(vs) => {
-                let (metric, matchers, shift) = vs_parts(vs)?;
+                let (metric, matchers, shift) = vs_parts(vs, self.query_range_ms)?;
                 self.instant_source(metric, matchers, shift)
             }
             Expr::MatrixSelector(ms) => {
-                let (metric, matchers, shift) = vs_parts(&ms.vs)?;
+                let (metric, matchers, shift) = vs_parts(&ms.vs, self.query_range_ms)?;
                 Ok(Unresolved::TimeRange {
                     range: ms.range,
                     kind: TimeRangeKind::Range,
@@ -993,7 +998,7 @@ impl PromqlLowerer<'_> {
     fn lower_inner(&self, expr: &Expr) -> Result<Inner> {
         match expr {
             Expr::VectorSelector(vs) => {
-                let (metric, matchers, shift) = vs_parts(vs)?;
+                let (metric, matchers, shift) = vs_parts(vs, self.query_range_ms)?;
                 Ok(Inner {
                     metric,
                     matchers,
@@ -1003,7 +1008,7 @@ impl PromqlLowerer<'_> {
                 })
             }
             Expr::MatrixSelector(ms) => {
-                let (metric, matchers, shift) = vs_parts(&ms.vs)?;
+                let (metric, matchers, shift) = vs_parts(&ms.vs, self.query_range_ms)?;
                 Ok(Inner {
                     metric,
                     matchers,
@@ -1026,7 +1031,8 @@ impl PromqlLowerer<'_> {
                 call.func.name.to_string(),
             ));
         };
-        let (metric, matchers, window, shift) = extract_matrix(arg(call, matrix_idx)?)?;
+        let (metric, matchers, window, shift) =
+            extract_matrix(arg(call, matrix_idx)?, self.query_range_ms)?;
         Ok(Inner {
             metric,
             matchers,
@@ -1577,12 +1583,12 @@ fn clamp_bounds_filter(min: &Scalar, max: &Scalar, child: Unresolved) -> Unresol
 fn collect_metric_names(expr: &Expr, out: &mut Vec<String>) {
     match expr {
         Expr::VectorSelector(vs) => {
-            if let Ok((metric, ..)) = vs_parts(vs) {
+            if let Ok((metric, ..)) = vs_parts(vs, 0) {
                 out.push(metric);
             }
         }
         Expr::MatrixSelector(ms) => {
-            if let Ok((metric, ..)) = vs_parts(&ms.vs) {
+            if let Ok((metric, ..)) = vs_parts(&ms.vs, 0) {
                 out.push(metric);
             }
         }
@@ -1714,7 +1720,7 @@ fn resolve_group(agg: &AggregateExpr) -> Result<(Vec<ColumnRef>, bool)> {
 
 // ── Free helpers ──────────────────────────────────────────────────────────────
 
-fn vs_parts(vs: &VectorSelector) -> Result<(String, Vec<Scalar>, TimeShift)> {
+fn vs_parts(vs: &VectorSelector, query_range_ms: i64) -> Result<(String, Vec<Scalar>, TimeShift)> {
     // A non-equality `__name__` matcher (`=~` / `!~` / `!=`) selects *across*
     // metric names. `Source::TimeSeries { metric }` carries a single concrete
     // metric name, so there is no representation for a regex/negated name
@@ -1761,26 +1767,45 @@ fn vs_parts(vs: &VectorSelector) -> Result<(String, Vec<Scalar>, TimeShift)> {
         .collect();
     ms.sort_by(|a, b| a.name.cmp(&b.name).then_with(|| a.value.cmp(&b.value)));
     let matchers = ms.into_iter().map(matcher_to_compare).collect();
-    let shift = time_shift(vs.offset.as_ref(), vs.at.as_ref())?;
+    let shift = time_shift(vs.offset.as_ref(), vs.at.as_ref(), query_range_ms)?;
     Ok((metric, matchers, shift))
 }
 
 /// Convert the parser's `offset` / `@` modifiers into a [`TimeShift`] (issue
 /// #40). Offset is signed milliseconds; `@ <ts>` (parser seconds → ms) becomes
-/// an absolute anchor, `@ start()`/`@ end()` the range bounds.
-fn time_shift(offset: Option<&Offset>, at: Option<&ParserAtModifier>) -> Result<TimeShift> {
+/// an absolute anchor and `@ end()` the end of the query range. `@ start()`
+/// is `@ end()` with `query_range_ms` (D) more offset: the range starts D
+/// before it ends.
+fn time_shift(
+    offset: Option<&Offset>,
+    at: Option<&ParserAtModifier>,
+    query_range_ms: i64,
+) -> Result<TimeShift> {
     let offset_ms = match offset {
         None => 0,
         Some(Offset::Pos(d)) => duration_ms(*d)?,
         Some(Offset::Neg(d)) => -duration_ms(*d)?,
     };
-    let at = match at {
-        None => None,
-        Some(ParserAtModifier::Start) => Some(AtModifier::Start),
-        Some(ParserAtModifier::End) => Some(AtModifier::End),
-        Some(ParserAtModifier::At(t)) => Some(AtModifier::Timestamp(system_time_ms(*t)?)),
-    };
-    Ok(TimeShift { offset_ms, at })
+    Ok(match at {
+        None => TimeShift {
+            offset_ms,
+            at: None,
+        },
+        Some(ParserAtModifier::End) => TimeShift {
+            offset_ms,
+            at: Some(AtModifier::End),
+        },
+        Some(ParserAtModifier::Start) => TimeShift {
+            offset_ms: offset_ms.checked_add(query_range_ms).ok_or_else(|| {
+                LoweringError::InvalidParameter("offset plus query range overflows i64 ms".into())
+            })?,
+            at: Some(AtModifier::End),
+        },
+        Some(ParserAtModifier::At(t)) => TimeShift {
+            offset_ms,
+            at: Some(AtModifier::Timestamp(system_time_ms(*t)?)),
+        },
+    })
 }
 
 /// A `Duration` as `i64` milliseconds, rejecting an overflow rather than
@@ -1821,13 +1846,16 @@ fn matcher_to_compare(m: &Matcher) -> Scalar {
     }
 }
 
-fn extract_matrix(expr: &Expr) -> Result<(String, Vec<Scalar>, Duration, TimeShift)> {
+fn extract_matrix(
+    expr: &Expr,
+    query_range_ms: i64,
+) -> Result<(String, Vec<Scalar>, Duration, TimeShift)> {
     match expr {
         Expr::MatrixSelector(ms) => {
-            let (metric, matchers, shift) = vs_parts(&ms.vs)?;
+            let (metric, matchers, shift) = vs_parts(&ms.vs, query_range_ms)?;
             Ok((metric, matchers, ms.range, shift))
         }
-        Expr::Paren(p) => extract_matrix(&p.expr),
+        Expr::Paren(p) => extract_matrix(&p.expr, query_range_ms),
         // A range-vector function argument must be a (parenthesised) matrix
         // selector. Do NOT descend through an arbitrary `Call` — that would
         // silently strip an unsupported wrapper (`rate(deriv(m[5m]))` lowering

crates/frontend-promql/tests/promql_conformance.rs

@@ -1074,7 +1074,9 @@ fn at_modifier_lowers_to_a_time_shift() {
     let NonASAPOp::TimeShift { shift, .. } = child.expect_non_asap() else {
         panic!("expected a TimeShift for `@ start()`");
     };
-    assert_eq!(shift.at, Some(AtModifier::Start));
+    // An instant query's range starts where it ends: `@ end()`.
+    assert_eq!(shift.at, Some(AtModifier::End));
+    assert_eq!(shift.offset_ms, 0);
 
     // Offset and `@` compose: `@ end() offset 5m` carries both.
     let qe = ok("http_requests_total @ end() offset 5m");

crates/frontend-sql/src/sql/expr.rs

@@ -1,5 +1,6 @@
 use std::rc::Rc;
 
+use datafusion::arrow::datatypes::{DataType as ArrowDataType, TimeUnit};
 use datafusion::logical_expr::utils::{split_binary, split_conjunction};
 use datafusion::logical_expr::{BinaryExpr, Expr, Operator};
 
@@ -137,17 +138,27 @@ impl SqlLowerer<'_> {
 
             Expr::IsNotNull(inner) => Ok(Unresolved::IsNotNull(bx(inner)?)),
 
-            // Fold a numeric cast of a literal, as DataFusion's constant folding
-            // does: type coercion turns `latency < 100` on a DOUBLE column into
-            // `latency < CAST(100 AS DOUBLE)`, which lowers as `latency < 100.0`.
+            // Fold a numeric or timestamp cast of a literal, as DataFusion's
+            // constant folding does: type coercion turns `latency < 100` on a
+            // DOUBLE column into `latency < CAST(100 AS DOUBLE)`, which lowers
+            // as `latency < 100.0`, and `ts >= '2026-11-01'` into a cast of the
+            // string, which lowers as a `Timestamp` literal (ms since the epoch).
             Expr::Cast(c)
                 if matches!(c.expr.as_ref(), Expr::Literal(v, _) if !v.is_null())
-                    && (c.field.data_type().is_integer() || c.field.data_type().is_floating()) =>
+                    && (c.field.data_type().is_integer()
+                        || c.field.data_type().is_floating()
+                        || matches!(c.field.data_type(), ArrowDataType::Timestamp(..))) =>
             {
                 let Expr::Literal(sv, _) = c.expr.as_ref() else {
                     unreachable!()
                 };
-                match sv.cast_to(c.field.data_type()) {
+                let to = match c.field.data_type() {
+                    ArrowDataType::Timestamp(..) => {
+                        ArrowDataType::Timestamp(TimeUnit::Millisecond, None)
+                    }
+                    other => other.clone(),
+                };
+                match sv.cast_to(&to) {
                     Ok(folded) => scalar_value_to_asap(&folded).map(Unresolved::Literal),
                     Err(_) => Ok(Unresolved::Cast {
                         expr: bx(&c.expr)?,

crates/frontend-sql/src/sql/types.rs

@@ -53,6 +53,13 @@ pub(super) fn scalar_value_to_asap(sv: &DfScalarValue) -> Result<ScalarValue, Lo
         | DfScalarValue::LargeUtf8(Some(s))
         | DfScalarValue::Utf8View(Some(s)) => Ok(ScalarValue::Utf8(s.clone())),
         DfScalarValue::Boolean(Some(b)) => Ok(ScalarValue::Boolean(*b)),
+        // A timestamp is ms since the epoch, as the executor's columns.
+        DfScalarValue::TimestampSecond(Some(s), _) => Ok(ScalarValue::Timestamp(s * 1_000)),
+        DfScalarValue::TimestampMillisecond(Some(ms), _) => Ok(ScalarValue::Timestamp(*ms)),
+        DfScalarValue::TimestampMicrosecond(Some(us), _) => Ok(ScalarValue::Timestamp(us / 1_000)),
+        DfScalarValue::TimestampNanosecond(Some(ns), _) => {
+            Ok(ScalarValue::Timestamp(ns / 1_000_000))
+        }
         // All three of DataFusion's interval scalars land on one canonical
         // shape; the narrower two simply leave the fields they do not carry
         // at zero.