fix(tesseract): stop a rolling window's plan fanning out across measures

rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/pre_aggregations_compiler.rs

@@ -26,6 +26,7 @@ use cubenativeutils::CubeError;
 use cubenativeutils::CubeErrorCauseType;
 use itertools::Itertools;
 use std::collections::HashMap;
+use std::collections::HashSet;
 use std::fmt::Debug;
 use std::rc::Rc;
 
@@ -750,6 +751,38 @@ impl PreAggregationsCompiler {
         Ok(result)
     }
 
+    /// The measures each pre-aggregation of these cubes declares, by full name.
+    /// Read off the declarations alone — no source, join or union is built — so
+    /// a caller only asking which measures a rollup groups together does not
+    /// pay for compiling it.
+    ///
+    /// A `rollupLambda` declares none of its own; the rollups it unions are
+    /// themselves pre-aggregations of the cube and are listed in their own
+    /// right.
+    pub fn declared_measures(
+        query_tools: Rc<State>,
+        cube_names: &Vec<String>,
+    ) -> Result<Vec<HashSet<String>>, CubeError> {
+        let mut result = Vec::new();
+        for cube_name in cube_names.iter() {
+            let pre_aggregations = query_tools
+                .cube_evaluator()
+                .pre_aggregations_for_cube_as_array(cube_name.clone())?;
+            for pre_aggregation in pre_aggregations.iter() {
+                let Some(refs) = pre_aggregation.measure_references()? else {
+                    continue;
+                };
+                let name = PreAggregationFullName::new(
+                    cube_name.clone(),
+                    pre_aggregation.static_data().name.clone(),
+                );
+                let symbols = Self::symbols_from_ref(query_tools.clone(), &name, refs, |_| Ok(()))?;
+                result.push(symbols.iter().map(|s| s.full_name()).collect());
+            }
+        }
+        Ok(result)
+    }
+
     pub fn compile_origin_sql_pre_aggregation(
         &mut self,
         cube_name: &String,

rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/filter/operators/filter_sql_context.rs

@@ -181,6 +181,27 @@ impl<'a> FilterSqlContext<'a> {
         self.plan_templates.convert_tz(field.to_string())
     }
 
+    /// The rolling window's series bounds as literal parameters, or `None`
+    /// when they can only be read back off the series itself.
+    ///
+    /// Raw values are spliced into pre-aggregation SQL verbatim rather than
+    /// allocated as parameters, which a bound cannot be rendered as.
+    pub fn date_range_literals(
+        &self,
+        range: &Option<(String, String)>,
+    ) -> Result<Option<(String, String)>, CubeError> {
+        let Some((from, to)) = range else {
+            return Ok(None);
+        };
+        if self.use_raw_values {
+            return Ok(None);
+        }
+        Ok(Some((
+            self.format_and_allocate_from_date(from)?,
+            self.format_and_allocate_to_date(to)?,
+        )))
+    }
+
     pub fn date_range_from_time_series(&self) -> Result<(String, String), CubeError> {
         Ok((
             self.time_series_bound("min", "date_from")?,

rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/filter/operators/rolling_window.rs

@@ -6,7 +6,10 @@ use cubenativeutils::CubeError;
 
 impl FilterOperationSql for RegularRollingWindowOp {
     fn to_sql(&self, ctx: &FilterSqlContext) -> Result<String, CubeError> {
-        let (from, to) = ctx.date_range_from_time_series()?;
+        let (from, to) = match ctx.date_range_literals(&self.series_range)? {
+            Some(range) => range,
+            None => ctx.date_range_from_time_series()?,
+        };
 
         let from = ctx.extend_date_range_bound(from, &self.trailing, true)?;
         let to = ctx.extend_date_range_bound(to, &self.leading, false)?;

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/filter/operators/rolling_window.rs

@@ -1,15 +1,30 @@
 /// `RegularRollingWindow` filter operation: trailing and leading
 /// interval bounds of a rolling window relative to each time-series
 /// point.
+///
+/// `series_range` holds the outer bounds of the series the window walks, when
+/// they are known at plan time. The filter restricts the base scan to the span
+/// the widest window can reach, and a literal span is one an engine can
+/// eliminate partitions by; without it the bounds are read back off the series
+/// with a scalar sub-select, which is opaque to pruning.
 #[derive(Clone, Debug)]
 pub struct RegularRollingWindowOp {
     pub(crate) trailing: Option<String>,
     pub(crate) leading: Option<String>,
+    pub(crate) series_range: Option<(String, String)>,
 }
 
 impl RegularRollingWindowOp {
-    pub fn new(trailing: Option<String>, leading: Option<String>) -> Self {
-        Self { trailing, leading }
+    pub fn new(
+        trailing: Option<String>,
+        leading: Option<String>,
+        series_range: Option<(String, String)>,
+    ) -> Self {
+        Self {
+            trailing,
+            leading,
+            series_range,
+        }
     }
 }
 

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/filter/typed_filter.rs

@@ -284,7 +284,15 @@ impl TypedFilterBuilder {
                 FilterOperator::RegularRollingWindowDateRange => {
                     let trailing = values.get(2).and_then(|v| v.to_param_string());
                     let leading = values.get(3).and_then(|v| v.to_param_string());
-                    FilterOp::RegularRollingWindow(RegularRollingWindowOp::new(trailing, leading))
+                    let series_range = values
+                        .get(4)
+                        .and_then(|v| v.to_param_string())
+                        .zip(values.get(5).and_then(|v| v.to_param_string()));
+                    FilterOp::RegularRollingWindow(RegularRollingWindowOp::new(
+                        trailing,
+                        leading,
+                        series_range,
+                    ))
                 }
                 FilterOperator::RollingWindowOffsetDateRange => {
                     let from = values.first().and_then(|v| v.to_param_string());

rust/cube/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/member_query_planner.rs

@@ -516,6 +516,18 @@ impl MultiStageMemberQueryPlanner {
             self.description.member_node().clone()
         };
         let member_node = &member_node;
+        let co_measures = self
+            .description
+            .co_measures()
+            .into_iter()
+            .map(|measure| {
+                if leaf_as_state {
+                    transforms::measures_as_state(&measure)
+                } else {
+                    Ok(measure)
+                }
+            })
+            .collect::<Result<Vec<_>, _>>()?;
         let mut dimensions = self.description.state().dimensions().clone();
         let mut time_dimensions = self.description.state().time_dimensions().clone();
         let mut measures = vec![];
@@ -541,6 +553,7 @@ impl MultiStageMemberQueryPlanner {
                 _ => {}
             }
         }
+        measures.extend(co_measures.iter().cloned());
 
         let mut measures_filters = self.description.state().measures_filters().clone();
         if leaf_as_state {
@@ -582,7 +595,9 @@ impl MultiStageMemberQueryPlanner {
             query_planner.plan(scope)
         })?;
         let leaf_measure_plan = MultiStageLeafMeasure {
-            measures: vec![member_node.clone()],
+            measures: std::iter::once(member_node.clone())
+                .chain(co_measures)
+                .collect_vec(),
             query,
             evaluation_context,
         };
@@ -630,12 +645,13 @@ impl MultiStageMemberQueryPlanner {
             .input()
             .iter()
             .map(|d| {
+                let measures = d.measures();
                 let schema = LogicalSchema::default()
                     .set_time_dimensions(d.state().time_dimensions().clone())
                     .set_dimensions(d.state().dimensions().clone())
-                    .set_measures(vec![d.member_node().clone()])
+                    .set_measures(measures.clone())
                     .into_rc();
-                (d.alias().clone(), vec![d.member_node().clone()], schema)
+                (d.alias().clone(), measures, schema)
             })
             .unique_by(|(a, _, _)| a.clone())
             .collect_vec()