multi stage rolling windows

Author: waralexrom

packages/cubejs-schema-compiler/test/integration/postgres/sql-generation.test.ts

@@ -155,6 +155,7 @@ describe('SQL Generation', () => {
         revenue_qtd_proxy: {
           type: 'sum',
           sql: \`\${revenue}\`,
+          multi_stage: true,
           rollingWindow: {
             type: 'to_date',
             granularity: 'quarter'
@@ -173,6 +174,7 @@ describe('SQL Generation', () => {
         revenueRollingDayAgo: {
           type: 'sum',
           sql: \`\${revenue_day_ago}\`,
+          multi_stage: true,
           rollingWindow: {
             trailing: '2 day',
             offset: 'start'

rust/cubesqlplanner/cubesqlplanner/src/planner/planners/multi_stage/multi_stage_query_planner.rs

@@ -281,6 +281,24 @@ impl MultiStageQueryPlanner {
                         granularity: None,
                     }
                 };
+
+                if !measure.is_multi_stage() {
+                    let childs = member_childs(&member, true)?;
+                    let measures = childs
+                        .iter()
+                        .filter(|s| s.as_measure().is_ok())
+                        .collect_vec();
+                    if !measures.is_empty() {
+                        return Err(CubeError::user(
+                            format!("Measure {} and references another measures ({}). In this case, {} must have multi_stage: true defined",
+                            member.full_name(),
+                            measures.into_iter().map(|m| m.full_name()).join(", "),
+                            member.full_name(),
+                                        ),
+                        ));
+                    }
+                }
+
                 let ungrouped = measure.is_rolling_window() && !measure.is_addictive();
 
                 let mut time_dimensions = self.query_properties.time_dimensions().clone();
@@ -323,14 +341,12 @@ impl MultiStageQueryPlanner {
                     &rolling_window,
                     state.clone(),
                 )?;
-                let base_member = MemberSymbol::new_measure(
-                    measure.new_unrolling(member.get_dependencies().is_empty()),
-                );
+                let base_member = MemberSymbol::new_measure(measure.new_unrolling());
 
                 let time_series =
                     self.add_time_series(time_dimension.clone(), state.clone(), descriptions)?;
 
-                let rolling_base = if base_member.get_dependencies().is_empty() {
+                let rolling_base = if !measure.is_multi_stage() {
                     self.add_rolling_window_base(
                         base_member,
                         base_rolling_state,

rust/cubesqlplanner/cubesqlplanner/src/planner/sql_evaluator/symbols/measure_symbol.rs

@@ -121,32 +121,36 @@ impl MeasureSymbol {
         })
     }
 
-    pub fn new_unrolling(&self, keep_type: bool) -> Rc<Self> {
-        let measure_type = if keep_type {
-            self.measure_type.clone()
+    pub fn new_unrolling(&self) -> Rc<Self> {
+        if self.is_rolling_window() {
+            let measure_type = if self.is_multi_stage {
+                format!("number")
+            } else {
+                self.measure_type.clone()
+            };
+            Rc::new(Self {
+                cube_name: self.cube_name.clone(),
+                name: self.name.clone(),
+                owned_by_cube: self.owned_by_cube,
+                measure_type,
+                rolling_window: None,
+                is_multi_stage: false,
+                is_reference: false,
+                is_view: self.is_view,
+                measure_filters: self.measure_filters.clone(),
+                measure_drill_filters: self.measure_drill_filters.clone(),
+                time_shift: self.time_shift.clone(),
+                measure_order_by: self.measure_order_by.clone(),
+                reduce_by: self.reduce_by.clone(),
+                add_group_by: self.add_group_by.clone(),
+                group_by: self.group_by.clone(),
+                member_sql: self.member_sql.clone(),
+                pk_sqls: self.pk_sqls.clone(),
+                is_splitted_source: self.is_splitted_source,
+            })
         } else {
-            format!("number")
-        };
-        Rc::new(Self {
-            cube_name: self.cube_name.clone(),
-            name: self.name.clone(),
-            owned_by_cube: self.owned_by_cube,
-            measure_type,
-            rolling_window: None,
-            is_multi_stage: self.is_multi_stage,
-            is_reference: self.is_reference,
-            is_view: self.is_view,
-            measure_filters: self.measure_filters.clone(),
-            measure_drill_filters: self.measure_drill_filters.clone(),
-            time_shift: self.time_shift.clone(),
-            measure_order_by: self.measure_order_by.clone(),
-            reduce_by: self.reduce_by.clone(),
-            add_group_by: self.add_group_by.clone(),
-            group_by: self.group_by.clone(),
-            member_sql: self.member_sql.clone(),
-            pk_sqls: self.pk_sqls.clone(),
-            is_splitted_source: self.is_splitted_source,
-        })
+            Rc::new(self.clone())
+        }
     }
 
     pub fn new_patched(